#!/usr/bin/env python3 """ Sweden 2005 Inheritance Tax Abolition — Difference-in-Differences ================================================================= Tests whether Sweden's 2005 abolition of the inheritance tax increased wealth concentration relative to Nordic controls (Denmark, Norway, Finland) that maintained inheritance taxes. Lean theorem link: dispersion_dominance — without tax incentive to disperse, concentration should increase. DID specification: top1_share_{c,t} = alpha_c + delta_t + beta * (Sweden_c x Post2005_t) + eps Prediction: beta > 0 (Swedish concentration increased relative to controls after abolition). Migrated from Fair Inheritance paper (Paper 9), Test 3. Review Notes (2026-03-12, v2): Theory match: CORRECT but INDIRECT. The Lean theorem (dispersion_dominance, PROVED) states concentrationWelfareCost ρ N > 0 when ρ < 1 and N ≥ 2 — dispersing capital among N heirs raises curvature. The test checks whether removing the tax incentive to disperse led to more concentration. The causal chain is indirect: remove tax → less dispersion → more concentration → (theorem) less output. The test only checks the concentration step. I/O audit: CORRECT. DID specification is standard; sign convention correct (positive beta = more concentration after abolition). Data-signal alignment: WEAK. Five critical problems: (1) PARALLEL TRENDS FAIL: Sweden (p=5e-6) and Norway (p=0.022). (2) EVENT STUDY NEGATIVE: After pre-trend adjustment, mean post-β = -0.012. Sweden concentration DECELERATED after abolition, opposite to prediction. (3) EXPANDED SCM NEGATIVE: With 11 OECD donors, gap = -0.004 (wrong sign). (4) NORWAY COUNTER-EVIDENCE: β=-0.014, p=0.011, wrong sign. (5) ROBUST TEST NULL: Excluding Norway, truncating to 2013: β=+0.005, p=0.52. Code quality: CLEAN (v2). 8 subtests, all with proper descriptions and caveats. v2 removed spurious top-10 CONSISTENT and uninformative simple means; added event study; expanded SCM from 3 Nordic to 11 OECD donors. Confidence in verdict: LOW. Weight of evidence is against the prediction. 1 CONSISTENT (tax revenue only), 3 AMBIGUOUS (null), 4 INCONSISTENT. Known limitations: (1) Only 3-4 countries → HC1 SEs unreliable. (2) WID data includes estimated values. (3) Many confounders. (4) Serial correlation. (5) Theorem is about estate-level dispersion; test measures aggregate shares. Date of review: 2026-03-12. """ import warnings import numpy as np import pandas as pd from scipy import stats from common.config import COMPILED_DIR from common.data import load_compiled from common.plotting import setup_figure, setup_subplots, save_figure from common.stats import verdict, log_result warnings.filterwarnings('ignore') # Treatment parameters TREATMENT_YEAR = 2005 PRE_START = 1995 PRE_END = 2004 POST_START = 2006 POST_END = 2020 # Norway abolished inheritance tax 2014-01-01 — truncate post-period for robustness POST_END_ROBUST = 2013 # Country codes (ISO3) TREATMENT = 'SWE' CONTROLS = ['DNK', 'NOR', 'FIN'] # Norway is only valid as control through 2013 CONTROLS_EXCL_NOR = ['DNK', 'FIN'] ALL_COUNTRIES = [TREATMENT] + CONTROLS # Expanded donor pool: OECD countries that maintained inheritance taxes 1995-2020 # Excludes: NOR (abolished 2014), AUT (abolished 2008), AUS/CAN/NZL (abolished pre-1995) # CHE excluded (complex cantonal system), ITA excluded (abolished 2001, reinstated 2006) SCM_DONORS = ['DNK', 'FIN', 'FRA', 'DEU', 'GBR', 'NLD', 'USA', 'JPN', 'BEL', 'ESP', 'KOR'] _ALL_SUBTESTS = [ 'sweden_did_top1', 'sweden_did_parallel_trends', 'sweden_did_event_study', 'sweden_did_synthetic_control', 'sweden_did_robust_excl_norway', 'sweden_tax_revenue_confirmation', 'norway_did_top1', 'norway_did_parallel_trends', ] _cache = {} def _load_wealth_shares(): """Load WID wealth share data.""" try: df = load_compiled('wid_wealth_shares') except FileNotFoundError: return None # Standardise column names df.columns = [c.strip().lower().replace(' ', '_') for c in df.columns] if 'country_code' not in df.columns or 'year' not in df.columns: return None df['year'] = pd.to_numeric(df['year'], errors='coerce') for col in ['top1_wealth_share', 'top10_wealth_share']: if col in df.columns: df[col] = pd.to_numeric(df[col], errors='coerce') return df def _load_tax_revenue(): """Load OECD inheritance tax revenue data.""" try: df = load_compiled('oecd_inheritance_tax_revenue') except FileNotFoundError: return None df.columns = [c.strip().lower().replace(' ', '_') for c in df.columns] if 'country_code' not in df.columns or 'year' not in df.columns: return None df['year'] = pd.to_numeric(df['year'], errors='coerce') if 'tax_revenue_pct_gdp' in df.columns: df['tax_revenue_pct_gdp'] = pd.to_numeric(df['tax_revenue_pct_gdp'], errors='coerce') return df def _build_did_panel(wealth_df, outcome_col='top1_wealth_share'): """ Build DID panel from wealth share data. Returns DataFrame with columns: country_code, year, outcome, treated, post, interaction or None if insufficient data. """ sub = wealth_df[wealth_df['country_code'].isin(ALL_COUNTRIES)].copy() sub = sub[sub['year'].between(PRE_START, POST_END)] sub = sub.dropna(subset=[outcome_col]) # Exclude treatment year itself sub = sub[sub['year'] != TREATMENT_YEAR] if len(sub) == 0: return None sub['outcome'] = sub[outcome_col] sub['treated'] = (sub['country_code'] == TREATMENT).astype(int) sub['post'] = (sub['year'] >= POST_START).astype(int) sub['interaction'] = sub['treated'] * sub['post'] # Check minimum data: need at least some observations in each cell cells = sub.groupby(['treated', 'post']).size() if len(cells) < 4 or cells.min() < 2: return None return sub def _run_did_ols(panel, test_name, outcome_label, controls_label='DK, NO, FI'): """ Run DID via OLS with country and year fixed effects. Returns a log_result dict. """ desc = ( f"Difference-in-differences: Sweden vs Nordic controls ({controls_label}) " f"around 2005 inheritance tax abolition. Outcome: {outcome_label}. " f"OLS with country and year FE, HC1 SEs (only {panel['country_code'].nunique()} " f"clusters — too few for reliable cluster-robust inference). " f"Positive beta means Swedish concentration increased relative to controls. " f"CAVEAT: parallel trends assumption fails (p < 1e-5); Norway abolished " f"its inheritance tax in 2014, contaminating the control group post-2013." ) try: import statsmodels.api as sm except ImportError: return log_result(test_name, np.nan, np.nan, 'SKIPPED', details={'reason': 'statsmodels not available'}, description=desc) df = panel.copy() # Country fixed effects (dummies, drop one) country_dummies = pd.get_dummies(df['country_code'], prefix='fe_c', drop_first=True, dtype=float) # Year fixed effects (dummies, drop one) year_dummies = pd.get_dummies(df['year'], prefix='fe_y', drop_first=True, dtype=float) X = pd.concat([df[['interaction']], country_dummies, year_dummies], axis=1) X = sm.add_constant(X) y = df['outcome'] try: # Cluster on country (small number of clusters, use HC1 as fallback) model = sm.OLS(y, X).fit(cov_type='HC1') beta = model.params['interaction'] p_val = model.pvalues['interaction'] ci_lo = model.conf_int().loc['interaction', 0] ci_hi = model.conf_int().loc['interaction', 1] except Exception as e: return log_result(test_name, np.nan, np.nan, 'SKIPPED', details={'reason': f'OLS failed: {e}'}, description=desc) sign_ok = beta > 0 # Prediction: abolition increases concentration verd = verdict(p_val, sign_ok, threshold=0.10) return log_result( test_name, beta, p_val, verd, details={ 'n_obs': len(df), 'n_countries': df['country_code'].nunique(), 'n_years': df['year'].nunique(), 'r2': round(model.rsquared, 4), 'ci_95': f'[{ci_lo:.4f}, {ci_hi:.4f}]', 'pre_mean_treated': round( df.loc[(df['treated'] == 1) & (df['post'] == 0), 'outcome'].mean(), 4), 'pre_mean_control': round( df.loc[(df['treated'] == 0) & (df['post'] == 0), 'outcome'].mean(), 4), 'post_mean_treated': round( df.loc[(df['treated'] == 1) & (df['post'] == 1), 'outcome'].mean(), 4), 'post_mean_control': round( df.loc[(df['treated'] == 0) & (df['post'] == 1), 'outcome'].mean(), 4), }, description=desc ) def _parallel_trends_test(panel): """ Test parallel trends assumption in pre-treatment period. Regress outcome on treated x year_trend interaction in pre-period. If coefficient is insignificant, parallel trends holds. Returns a log_result dict. """ desc = ( "Parallel trends check: regress pre-treatment wealth share on " "treated x year_trend interaction. Insignificant coefficient (p > 0.10) " "supports the parallel trends assumption required for DID validity. " "WARNING: Sweden's top-1% share trends up at +0.49pp/year pre-2005 " "while Denmark trends down at -0.33pp/year. This differential pre-trend " "(p < 1e-5) invalidates the standard DID identifying assumption." ) try: import statsmodels.api as sm except ImportError: return log_result('sweden_did_parallel_trends', np.nan, np.nan, 'SKIPPED', details={'reason': 'statsmodels not available'}, description=desc) pre = panel[panel['post'] == 0].copy() if len(pre) < 10: return log_result('sweden_did_parallel_trends', np.nan, np.nan, 'SKIPPED', details={'reason': f'Insufficient pre-period obs: {len(pre)}'}, description=desc) pre['year_trend'] = pre['year'] - PRE_START pre['treated_trend'] = pre['treated'] * pre['year_trend'] country_dummies = pd.get_dummies(pre['country_code'], prefix='fe_c', drop_first=True, dtype=float) X = pd.concat([pre[['year_trend', 'treated', 'treated_trend']], country_dummies], axis=1) X = sm.add_constant(X) y = pre['outcome'] try: model = sm.OLS(y, X).fit(cov_type='HC1') beta_trend = model.params['treated_trend'] p_trend = model.pvalues['treated_trend'] except Exception as e: return log_result('sweden_did_parallel_trends', np.nan, np.nan, 'SKIPPED', details={'reason': f'OLS failed: {e}'}, description=desc) # Parallel trends holds if treated_trend is NOT significant trends_ok = p_trend > 0.10 verd = 'CONSISTENT' if trends_ok else 'INCONSISTENT' return log_result( 'sweden_did_parallel_trends', beta_trend, p_trend, verd, details={ 'n_obs_pre': len(pre), 'interpretation': 'Parallel trends supported' if trends_ok else 'Differential pre-trends detected', }, description=desc ) def _synthetic_control(wealth_df, outcome_col='top1_wealth_share', donors=None): """ Synthetic control method: construct synthetic Sweden from donor pool by matching pre-treatment outcomes. Compare post-treatment divergence. Uses scipy.optimize to find non-negative weights that minimize pre-treatment MSPE. Returns a log_result dict and the synthetic series for plotting. """ desc = ( "Synthetic control: construct synthetic Sweden from 11 OECD donor countries " "(DNK, FIN, FRA, DEU, GBR, NLD, USA, JPN, BEL, ESP, KOR) that maintained " "inheritance taxes throughout 1995-2020. Minimizes pre-treatment MSPE on " "top-1% wealth shares. P-value from placebo distribution (11 placebos). " "Expanded donor pool eliminates the Norway contamination problem (prior " "version had 100% weight on Norway, which also abolished in 2014)." ) if donors is None: donors = SCM_DONORS try: from scipy.optimize import minimize except ImportError: return log_result('sweden_did_synthetic_control', np.nan, np.nan, 'SKIPPED', details={'reason': 'scipy.optimize not available'}, description=desc), None, None all_scm = [TREATMENT] + donors sub = wealth_df[wealth_df['country_code'].isin(all_scm)].copy() sub = sub.dropna(subset=[outcome_col]) # Build year x country pivot pivot = sub.pivot_table(index='year', columns='country_code', values=outcome_col) # Need at least some years in both periods for all countries pre_years = [y for y in range(PRE_START, PRE_END + 1) if y in pivot.index] post_years = [y for y in range(POST_START, POST_END + 1) if y in pivot.index] if len(pre_years) < 3 or len(post_years) < 3: return log_result('sweden_did_synthetic_control', np.nan, np.nan, 'SKIPPED', details={'reason': f'Insufficient years: pre={len(pre_years)}, post={len(post_years)}'}, description=desc), None, None # Check that treatment and at least 2 controls have data avail_controls = [c for c in donors if c in pivot.columns and pivot.loc[pre_years, c].notna().sum() >= 3] if TREATMENT not in pivot.columns or len(avail_controls) < 2: return log_result('sweden_did_synthetic_control', np.nan, np.nan, 'SKIPPED', details={'reason': 'Insufficient country coverage'}, description=desc), None, None # Interpolate small gaps pivot = pivot.interpolate(method='linear', limit=2) # Pre-treatment vectors Y_treat_pre = pivot.loc[pre_years, TREATMENT].values X_donors_pre = pivot.loc[pre_years, avail_controls].values # Drop years where treatment or any donor is NaN valid = ~np.isnan(Y_treat_pre) & ~np.any(np.isnan(X_donors_pre), axis=1) if valid.sum() < 3: return log_result('sweden_did_synthetic_control', np.nan, np.nan, 'SKIPPED', details={'reason': 'Too many NaN in pre-period'}, description=desc), None, None Y_treat_pre = Y_treat_pre[valid] X_donors_pre = X_donors_pre[valid] n_donors = len(avail_controls) # Minimize pre-treatment MSPE subject to weights summing to 1, non-negative def objective(w): synthetic = X_donors_pre @ w return np.mean((Y_treat_pre - synthetic) ** 2) constraints = {'type': 'eq', 'fun': lambda w: np.sum(w) - 1.0} bounds = [(0.0, 1.0)] * n_donors w0 = np.ones(n_donors) / n_donors result = minimize(objective, w0, method='SLSQP', bounds=bounds, constraints=constraints) if not result.success: return log_result('sweden_did_synthetic_control', np.nan, np.nan, 'SKIPPED', details={'reason': f'Optimization failed: {result.message}'}, description=desc), None, None w_star = result.x pre_mspe = result.fun # Construct full synthetic series all_years = sorted(set(pre_years + post_years)) Y_donors_full = pivot.loc[all_years, avail_controls].values Y_treat_full = pivot.loc[all_years, TREATMENT].values valid_full = ~np.isnan(Y_treat_full) & ~np.any(np.isnan(Y_donors_full), axis=1) years_valid = np.array(all_years)[valid_full] Y_treat_valid = Y_treat_full[valid_full] Y_synth_valid = Y_donors_full[valid_full] @ w_star # Post-treatment gap post_mask = years_valid >= POST_START if post_mask.sum() < 2: return log_result('sweden_did_synthetic_control', np.nan, np.nan, 'SKIPPED', details={'reason': 'Insufficient post-treatment synthetic obs'}, description=desc), None, None gap_post = (Y_treat_valid[post_mask] - Y_synth_valid[post_mask]).mean() # Placebo inference: run SCM for each control country placebo_gaps = [] for placebo_treat in avail_controls: placebo_donors = [c for c in avail_controls if c != placebo_treat] if len(placebo_donors) < 1: continue Y_pt_pre = pivot.loc[pre_years, placebo_treat].values X_pd_pre = pivot.loc[pre_years, placebo_donors].values pv = ~np.isnan(Y_pt_pre) & ~np.any(np.isnan(X_pd_pre), axis=1) if pv.sum() < 3: continue n_pd = len(placebo_donors) def obj_p(w, Y=Y_pt_pre[pv], X=X_pd_pre[pv]): return np.mean((Y - X @ w) ** 2) cons_p = {'type': 'eq', 'fun': lambda w: np.sum(w) - 1.0} bnd_p = [(0.0, 1.0)] * n_pd w0_p = np.ones(n_pd) / n_pd res_p = minimize(obj_p, w0_p, method='SLSQP', bounds=bnd_p, constraints=cons_p) if not res_p.success: continue # Placebo post gap Y_pt_full = pivot.loc[all_years, placebo_treat].values Y_pd_full = pivot.loc[all_years, placebo_donors].values pv_f = ~np.isnan(Y_pt_full) & ~np.any(np.isnan(Y_pd_full), axis=1) Y_synth_p = Y_pd_full[pv_f] @ res_p.x yrs_p = np.array(all_years)[pv_f] pm = yrs_p >= POST_START if pm.sum() >= 2: pg = (Y_pt_full[pv_f][pm] - Y_synth_p[pm]).mean() placebo_gaps.append(pg) # P-value: fraction of placebo gaps >= treatment gap if len(placebo_gaps) >= 2: p_val = np.mean([g >= gap_post for g in placebo_gaps]) # With small number of placebos, use rank-based p-value all_gaps = [gap_post] + placebo_gaps rank = sum(1 for g in all_gaps if g >= gap_post) p_val = rank / len(all_gaps) else: # Cannot compute placebo p-value; use pre-MSPE based heuristic p_val = np.nan sign_ok = gap_post > 0 if np.isnan(p_val): verd = 'AMBIGUOUS' if sign_ok else 'INCONSISTENT' else: verd = verdict(p_val, sign_ok, threshold=0.10) # Store for figures synth_series = pd.Series(Y_synth_valid, index=years_valid, name='synthetic_sweden') treat_series = pd.Series(Y_treat_valid, index=years_valid, name='actual_sweden') weights_dict = {c: round(w, 4) for c, w in zip(avail_controls, w_star)} return log_result( 'sweden_did_synthetic_control', gap_post, p_val, verd, details={ 'weights': weights_dict, 'pre_mspe': round(pre_mspe, 6), 'post_gap_mean': round(gap_post, 4), 'n_placebo': len(placebo_gaps), 'placebo_gaps': [round(g, 4) for g in placebo_gaps], }, description=desc ), treat_series, synth_series def _event_study(panel): """ Event study: estimate year-by-year treatment effects relative to t=-1 (2004). Pre-treatment coefficients should be ~0 if parallel trends hold. Post-treatment coefficients should be > 0 if abolition increased concentration. """ desc = ( "Event study: year-by-year DID treatment effects for Sweden vs Nordic " "controls (DK, NO, FI). OLS with country and year FE, HC1 SEs. " "Estimates beta_k for each year k relative to 2004 (omitted). " "Pre-treatment betas ~0 supports parallel trends; post-treatment " "betas > 0 supports concentration increase. Joint F-test on " "pre-treatment betas tests parallel trends. " "CAVEAT: only 4 clusters; HC1 SEs may be too narrow." ) try: import statsmodels.api as sm except ImportError: return log_result('sweden_did_event_study', np.nan, np.nan, 'SKIPPED', details={'reason': 'statsmodels not available'}, description=desc) df = panel.copy() base_year = 2004 # last pre-treatment year # Create year-relative-to-treatment indicators interacted with treated years = sorted(df['year'].unique()) event_cols = [] for y in years: if y == base_year: continue # omitted reference year col_name = f'event_{y}' df[col_name] = ((df['treated'] == 1) & (df['year'] == y)).astype(float) event_cols.append((y, col_name)) # Country and year FE country_dummies = pd.get_dummies(df['country_code'], prefix='fe_c', drop_first=True, dtype=float) year_dummies = pd.get_dummies(df['year'], prefix='fe_y', drop_first=True, dtype=float) X = pd.concat([df[[col for _, col in event_cols]], country_dummies, year_dummies], axis=1) X = sm.add_constant(X) y = df['outcome'] try: model = sm.OLS(y, X).fit(cov_type='HC1') except Exception as e: return log_result('sweden_did_event_study', np.nan, np.nan, 'SKIPPED', details={'reason': f'OLS failed: {e}'}, description=desc) # Extract event-study coefficients es_years = [] es_betas = [] es_ci_lo = [] es_ci_hi = [] es_pvals = [] for y_val, col in event_cols: if col in model.params.index: es_years.append(y_val - TREATMENT_YEAR) # relative to 2005 es_betas.append(model.params[col]) es_ci_lo.append(model.conf_int().loc[col, 0]) es_ci_hi.append(model.conf_int().loc[col, 1]) es_pvals.append(model.pvalues[col]) # Add the omitted base year (beta = 0, no CI) es_years.append(base_year - TREATMENT_YEAR) es_betas.append(0.0) es_ci_lo.append(0.0) es_ci_hi.append(0.0) es_pvals.append(1.0) # Sort by year sort_idx = np.argsort(es_years) es_years = [es_years[i] for i in sort_idx] es_betas = [es_betas[i] for i in sort_idx] es_ci_lo = [es_ci_lo[i] for i in sort_idx] es_ci_hi = [es_ci_hi[i] for i in sort_idx] es_pvals = [es_pvals[i] for i in sort_idx] # Joint F-test on pre-treatment betas (parallel trends test) pre_cols = [col for y_val, col in event_cols if y_val < TREATMENT_YEAR and col in model.params.index] if len(pre_cols) >= 2: restriction = ' = '.join([f'{col}' for col in pre_cols]) + ' = 0' try: f_test = model.f_test(restriction) f_stat = float(f_test.fvalue) f_pval = float(f_test.pvalue) except Exception: f_stat, f_pval = np.nan, np.nan else: f_stat, f_pval = np.nan, np.nan # Post-treatment mean beta post_betas = [b for yr, b in zip(es_years, es_betas) if yr > 0] mean_post_beta = np.mean(post_betas) if post_betas else np.nan # Verdict: parallel trends must hold (F-test p > 0.10) AND mean post beta > 0 # But we know parallel trends fail, so report honestly pre_trends_ok = f_pval > 0.10 if not np.isnan(f_pval) else False post_positive = mean_post_beta > 0 if pre_trends_ok and post_positive: verd = 'CONSISTENT' elif not pre_trends_ok: verd = 'INCONSISTENT' # parallel trends fail else: verd = 'AMBIGUOUS' # Cache for figure _cache['event_study'] = { 'years_relative': es_years, 'betas': es_betas, 'ci_lo': es_ci_lo, 'ci_hi': es_ci_hi, } return log_result( 'sweden_did_event_study', mean_post_beta, f_pval, verd, details={ 'mean_post_beta': round(mean_post_beta, 4), 'f_stat_pretrend': round(f_stat, 3) if not np.isnan(f_stat) else None, 'f_pval_pretrend': round(f_pval, 6) if not np.isnan(f_pval) else None, 'pre_trends_ok': pre_trends_ok, 'n_pre_years': len(pre_cols), 'n_post_years': len(post_betas), 'n_obs': len(df), }, description=desc) def main(): results = [] # Load wealth share data wealth_df = _load_wealth_shares() if wealth_df is None: skip_desc = ( "Sweden 2005 inheritance tax abolition DID: tests whether abolition " "increased top-1% wealth share relative to Nordic controls. " "Requires wid_wealth_shares.csv in data/compiled/." ) results.append(log_result( 'sweden_did_top1', np.nan, np.nan, 'SKIPPED', details={'reason': 'wid_wealth_shares.csv not found'}, description=skip_desc)) _cache['results'] = results return results # --- Test 1: DID on top 1% wealth share --- panel_top1 = _build_did_panel(wealth_df, 'top1_wealth_share') if panel_top1 is not None: results.append(_run_did_ols(panel_top1, 'sweden_did_top1', 'top 1% wealth share')) # Parallel trends check results.append(_parallel_trends_test(panel_top1)) _cache['panel_top1'] = panel_top1 else: results.append(log_result( 'sweden_did_top1', np.nan, np.nan, 'SKIPPED', details={'reason': 'Insufficient top1_wealth_share data for DID panel'}, description="DID on top 1% wealth share around Sweden 2005 abolition.")) # --- Test 2: Event study --- if panel_top1 is not None: results.append(_event_study(panel_top1)) # --- Test 3: Synthetic control --- scm_result, treat_series, synth_series = _synthetic_control(wealth_df, 'top1_wealth_share') results.append(scm_result) if treat_series is not None: _cache['scm_treat'] = treat_series _cache['scm_synth'] = synth_series # --- Test 4: Robustness DID excluding Norway, truncated post-period --- # Norway abolished its inheritance tax in 2014. Use only DNK+FIN as controls, # truncate post-period to 2006-2013 for a clean comparison. robust_df = wealth_df[ wealth_df['country_code'].isin([TREATMENT] + CONTROLS_EXCL_NOR) ].copy() robust_df = robust_df[robust_df['year'].between(PRE_START, POST_END_ROBUST)] robust_df = robust_df[robust_df['year'] != TREATMENT_YEAR] robust_df = robust_df.dropna(subset=['top1_wealth_share']) if len(robust_df) > 0: robust_df['outcome'] = robust_df['top1_wealth_share'] robust_df['treated'] = (robust_df['country_code'] == TREATMENT).astype(int) robust_df['post'] = (robust_df['year'] >= POST_START).astype(int) robust_df['interaction'] = robust_df['treated'] * robust_df['post'] cells = robust_df.groupby(['treated', 'post']).size() if len(cells) >= 4 and cells.min() >= 2: results.append(_run_did_ols( robust_df, 'sweden_did_robust_excl_norway', 'top 1% wealth share (excl Norway, post to 2013)', controls_label='DK, FI')) _cache['robust_panel'] = robust_df # --- Test 5: Tax revenue confirmation --- tax_df = _load_tax_revenue() desc_tax = ( "Confirms the treatment: Sweden inheritance tax revenue drops to ~0 " "after 2005 while controls maintain positive revenue. " "Data: OECD Revenue Statistics (1965-2022), inheritance/estate/gift tax " "revenue as % GDP. Sweden: 0.090 (2004) -> 0.004 (2006) -> 0.000 (2008+). " "DK and FI maintained 0.2-0.3% throughout." ) if tax_df is not None and 'tax_revenue_pct_gdp' in tax_df.columns: se_tax = tax_df[tax_df['country_code'] == TREATMENT] se_post = se_tax[se_tax['year'] >= POST_START]['tax_revenue_pct_gdp'] ctrl_tax = tax_df[tax_df['country_code'].isin(CONTROLS)] ctrl_post = ctrl_tax[ctrl_tax['year'] >= POST_START]['tax_revenue_pct_gdp'] if len(se_post) >= 2 and len(ctrl_post) >= 2: se_mean = se_post.mean() ctrl_mean = ctrl_post.mean() # Sweden should be near zero, controls should be positive revenue_dropped = se_mean < 0.05 and ctrl_mean > se_mean verd_tax = 'CONSISTENT' if revenue_dropped else 'INCONSISTENT' results.append(log_result( 'sweden_tax_revenue_confirmation', se_mean, np.nan, verd_tax, details={ 'sweden_post_mean': round(se_mean, 4), 'control_post_mean': round(ctrl_mean, 4), }, description=desc_tax)) else: results.append(log_result( 'sweden_tax_revenue_confirmation', np.nan, np.nan, 'SKIPPED', details={'reason': 'Insufficient post-period tax revenue data'}, description=desc_tax)) else: results.append(log_result( 'sweden_tax_revenue_confirmation', np.nan, np.nan, 'SKIPPED', details={'reason': 'oecd_inheritance_tax_revenue.csv not found'}, description=desc_tax)) _cache['tax_df'] = tax_df # --- Test 7: Norway 2014 inheritance tax abolition DID --- # Norway abolished its inheritance tax on January 1, 2014. This provides # a CLEANER natural experiment than Sweden 2005 because: # (a) Pre-period 2006-2013 avoids Sweden's abolition contaminating trends # (b) Controls (DNK, FIN) maintained inheritance taxes throughout # (c) Sweden (already abolished) is excluded from the control group nor_countries = ['NOR', 'DNK', 'FIN'] nor_sub = wealth_df[wealth_df['country_code'].isin(nor_countries)].copy() nor_sub = nor_sub[nor_sub['year'].between(2006, 2020)] nor_sub = nor_sub[nor_sub['year'] != 2014] nor_sub = nor_sub.dropna(subset=['top1_wealth_share']) nor_did_desc = ( "Difference-in-differences: Norway vs Nordic controls (DK, FI) " "around 2014 inheritance tax abolition. Outcome: top 1% wealth share. " "OLS with country and year FE, HC1 SEs. " "This is a CLEANER natural experiment than Sweden 2005: " "(a) pre-period 2006-2013 avoids Sweden contamination, " "(b) controls (DNK, FIN) maintained inheritance taxes throughout, " "(c) Sweden excluded from control group (already abolished 2005). " "Positive beta means Norwegian concentration increased relative to controls." ) if len(nor_sub) > 0 and 'top1_wealth_share' in nor_sub.columns: nor_sub['outcome'] = nor_sub['top1_wealth_share'] nor_sub['treated'] = (nor_sub['country_code'] == 'NOR').astype(int) nor_sub['post'] = (nor_sub['year'] >= 2015).astype(int) nor_sub['interaction'] = nor_sub['treated'] * nor_sub['post'] cells = nor_sub.groupby(['treated', 'post']).size() if len(cells) >= 4 and cells.min() >= 2: _cache['norway_panel'] = nor_sub # Run DID OLS try: import statsmodels.api as sm df_nor = nor_sub.copy() country_dummies = pd.get_dummies(df_nor['country_code'], prefix='fe_c', drop_first=True, dtype=float) year_dummies = pd.get_dummies(df_nor['year'], prefix='fe_y', drop_first=True, dtype=float) X_nor = pd.concat([df_nor[['interaction']], country_dummies, year_dummies], axis=1) X_nor = sm.add_constant(X_nor) y_nor = df_nor['outcome'] model_nor = sm.OLS(y_nor, X_nor).fit(cov_type='HC1') beta_nor = model_nor.params['interaction'] p_nor = model_nor.pvalues['interaction'] ci_lo_nor = model_nor.conf_int().loc['interaction', 0] ci_hi_nor = model_nor.conf_int().loc['interaction', 1] sign_ok_nor = beta_nor > 0 verd_nor = verdict(p_nor, sign_ok_nor, threshold=0.10) results.append(log_result( 'norway_did_top1', beta_nor, p_nor, verd_nor, details={ 'n_obs': len(df_nor), 'n_countries': df_nor['country_code'].nunique(), 'n_years': df_nor['year'].nunique(), 'r2': round(model_nor.rsquared, 4), 'ci_95': f'[{ci_lo_nor:.4f}, {ci_hi_nor:.4f}]', 'pre_mean_treated': round( df_nor.loc[(df_nor['treated'] == 1) & (df_nor['post'] == 0), 'outcome'].mean(), 4), 'pre_mean_control': round( df_nor.loc[(df_nor['treated'] == 0) & (df_nor['post'] == 0), 'outcome'].mean(), 4), 'post_mean_treated': round( df_nor.loc[(df_nor['treated'] == 1) & (df_nor['post'] == 1), 'outcome'].mean(), 4), 'post_mean_control': round( df_nor.loc[(df_nor['treated'] == 0) & (df_nor['post'] == 1), 'outcome'].mean(), 4), }, description=nor_did_desc)) except ImportError: results.append(log_result( 'norway_did_top1', np.nan, np.nan, 'SKIPPED', details={'reason': 'statsmodels not available'}, description=nor_did_desc)) except Exception as e: results.append(log_result( 'norway_did_top1', np.nan, np.nan, 'SKIPPED', details={'reason': f'OLS failed: {e}'}, description=nor_did_desc)) # --- Test 8: Norway parallel trends check --- nor_pt_desc = ( "Parallel trends check for Norway 2014 DID: regress pre-treatment " "(2006-2013) wealth share on treated x year_trend interaction. " "Insignificant coefficient (p > 0.10) supports parallel trends. " "This should be cleaner than the Sweden test which fails with p=5e-6." ) try: import statsmodels.api as sm pre_nor = nor_sub[nor_sub['post'] == 0].copy() if len(pre_nor) >= 10: pre_nor['year_trend'] = pre_nor['year'] - 2006 pre_nor['treated_trend'] = pre_nor['treated'] * pre_nor['year_trend'] country_dummies_pt = pd.get_dummies(pre_nor['country_code'], prefix='fe_c', drop_first=True, dtype=float) X_pt = pd.concat([pre_nor[['year_trend', 'treated', 'treated_trend']], country_dummies_pt], axis=1) X_pt = sm.add_constant(X_pt) y_pt = pre_nor['outcome'] model_pt = sm.OLS(y_pt, X_pt).fit(cov_type='HC1') beta_pt = model_pt.params['treated_trend'] p_pt = model_pt.pvalues['treated_trend'] trends_ok_nor = p_pt > 0.10 verd_pt = 'CONSISTENT' if trends_ok_nor else 'INCONSISTENT' results.append(log_result( 'norway_did_parallel_trends', beta_pt, p_pt, verd_pt, details={ 'n_obs_pre': len(pre_nor), 'interpretation': 'Parallel trends supported' if trends_ok_nor else 'Differential pre-trends detected', }, description=nor_pt_desc)) else: results.append(log_result( 'norway_did_parallel_trends', np.nan, np.nan, 'SKIPPED', details={'reason': f'Insufficient pre-period obs: {len(pre_nor)}'}, description=nor_pt_desc)) except ImportError: results.append(log_result( 'norway_did_parallel_trends', np.nan, np.nan, 'SKIPPED', details={'reason': 'statsmodels not available'}, description=nor_pt_desc)) except Exception as e: results.append(log_result( 'norway_did_parallel_trends', np.nan, np.nan, 'SKIPPED', details={'reason': f'Parallel trends test failed: {e}'}, description=nor_pt_desc)) else: results.append(log_result( 'norway_did_top1', np.nan, np.nan, 'SKIPPED', details={'reason': 'Insufficient cell counts for Norway DID'}, description=nor_did_desc)) else: results.append(log_result( 'norway_did_top1', np.nan, np.nan, 'SKIPPED', details={'reason': 'No top1_wealth_share data for NOR/DNK/FIN'}, description=nor_did_desc)) _cache['results'] = results _cache['wealth_df'] = wealth_df return results def figures(): """Generate per-subtest figures for Sweden/Norway inheritance tax DID.""" if not _cache: main() paths = [] wealth_df = _cache.get('wealth_df') if wealth_df is None: return paths colors = {'SWE': '#e74c3c', 'DNK': '#3498db', 'NOR': '#2ecc71', 'FIN': '#9b59b6'} # --- Figure 1: Sweden DID top 1% + parallel trends --- panel = _cache.get('panel_top1') if panel is not None and len(panel) > 0: fig, axes = setup_subplots(1, 2, width=14, height=6) ax = axes[0] for cc in ALL_COUNTRIES: sub = panel[panel['country_code'] == cc].sort_values('year') if len(sub) > 0: ax.plot(sub['year'], sub['outcome'], 'o-', color=colors.get(cc, 'gray'), label=cc, markersize=4, linewidth=1.5) ax.axvline(TREATMENT_YEAR, color='black', linestyle='--', alpha=0.7, label='Abolition (2005)') ax.set_xlabel('Year') ax.set_ylabel('Top 1% Wealth Share') ax.set_title('Wealth Concentration: Sweden vs Nordic Controls') ax.legend(fontsize=9) ax = axes[1] for label, name, color in [(1, 'Sweden (treated)', '#e74c3c'), (0, 'Nordic controls', '#3498db')]: grp = panel[panel['treated'] == label].groupby('year')['outcome'].mean() grp = grp.sort_index() ax.plot(grp.index, grp.values, 'o-', color=color, label=name, markersize=4, linewidth=1.5) ax.axvline(TREATMENT_YEAR, color='black', linestyle='--', alpha=0.7, label='Abolition (2005)') ax.set_xlabel('Year') ax.set_ylabel('Top 1% Wealth Share') ax.set_title('DID: Treatment vs Control Group Means') ax.legend(fontsize=9) fig.suptitle('Sweden 2005 Inheritance Tax Abolition — Top 1% DID', fontsize=13, y=1.02) paths += save_figure(fig, 'fair_inheritance_sweden_did_top1', subfolder='applications') # --- Figure 2: Event study --- es_data = _cache.get('event_study') if es_data is not None: fig, ax = setup_figure(width=10, height=6) years_rel = es_data['years_relative'] betas = es_data['betas'] ci_lo = es_data['ci_lo'] ci_hi = es_data['ci_hi'] ax.errorbar(years_rel, betas, yerr=[np.array(betas) - np.array(ci_lo), np.array(ci_hi) - np.array(betas)], fmt='o-', color='#e74c3c', capsize=3, markersize=5, linewidth=1.5) ax.axhline(0, color='black', linewidth=0.8, linestyle='-') ax.axvline(0, color='black', linewidth=0.8, linestyle='--', alpha=0.7, label='Abolition (2005)') ax.fill_between([-0.5, 0.5], ax.get_ylim()[0], ax.get_ylim()[1], alpha=0.1, color='gray') ax.set_xlabel('Years Relative to Abolition (2005)') ax.set_ylabel('Treatment Effect (beta_k)') ax.set_title('Event Study: Sweden DID Year-by-Year Treatment Effects') ax.legend(fontsize=9) paths += save_figure(fig, 'fair_inheritance_sweden_event_study', subfolder='applications') # --- Figure 3: Synthetic control --- treat_series = _cache.get('scm_treat') synth_series = _cache.get('scm_synth') if treat_series is not None and synth_series is not None: fig, axes = setup_subplots(1, 2, width=14, height=6) ax = axes[0] ax.plot(treat_series.index, treat_series.values, 'o-', color='#e74c3c', label='Actual Sweden', markersize=4, linewidth=1.5) ax.plot(synth_series.index, synth_series.values, 's--', color='#3498db', label='Synthetic Sweden', markersize=4, linewidth=1.5) ax.axvline(TREATMENT_YEAR, color='black', linestyle='--', alpha=0.7, label='Abolition (2005)') ax.set_xlabel('Year') ax.set_ylabel('Top 1% Wealth Share') ax.set_title('Synthetic Control: Actual vs Synthetic Sweden') ax.legend(fontsize=9) ax = axes[1] common_idx = treat_series.index.intersection(synth_series.index) gap = treat_series.loc[common_idx] - synth_series.loc[common_idx] ax.bar(gap.index, gap.values, color=['#e74c3c' if g > 0 else '#3498db' for g in gap.values], alpha=0.7) ax.axhline(0, color='black', linewidth=0.8) ax.axvline(TREATMENT_YEAR, color='black', linestyle='--', alpha=0.7, label='Abolition (2005)') ax.set_xlabel('Year') ax.set_ylabel('Gap (Actual - Synthetic)') ax.set_title('Treatment Effect: Post-Abolition Gap') ax.legend(fontsize=9) fig.suptitle('Synthetic Control Method — Sweden Inheritance Tax', fontsize=13, y=1.02) paths += save_figure(fig, 'fair_inheritance_sweden_scm', subfolder='applications') # --- Figure 4: Robust DID excluding Norway --- robust = _cache.get('robust_panel') if robust is not None and len(robust) > 0: fig, ax = setup_figure(width=8, height=5) for cc in ['SWE', 'DNK', 'FIN']: sub = robust[robust['country_code'] == cc].sort_values('year') if len(sub) > 0: ax.plot(sub['year'], sub['outcome'], 'o-', color=colors.get(cc, 'gray'), label=cc, markersize=4, linewidth=1.5) ax.axvline(TREATMENT_YEAR, color='black', linestyle='--', alpha=0.7, label='Abolition (2005)') ax.set_xlabel('Year') ax.set_ylabel('Top 1% Wealth Share') ax.set_title('Robust DID: Excluding Norway, Post to 2013') ax.legend(fontsize=9) paths += save_figure(fig, 'fair_inheritance_sweden_robust', subfolder='applications') # --- Figure 5: Tax revenue (even if SKIPPED, show available data) --- tax_df = _cache.get('tax_df') if tax_df is not None and 'tax_revenue_pct_gdp' in tax_df.columns: fig, ax = setup_figure(width=8, height=5) for cc in ['SWE', 'NOR', 'DNK', 'FIN']: sub = tax_df[tax_df['country_code'] == cc].sort_values('year') if len(sub) > 0: ax.plot(sub['year'], sub['tax_revenue_pct_gdp'], 'o-', color=colors.get(cc, 'gray'), label=cc, markersize=6) ax.set_xlabel('Year') ax.set_ylabel('Inheritance Tax Revenue (% GDP)') ax.set_title('OECD Inheritance Tax Revenue — Nordic Countries') ax.legend(fontsize=9) paths += save_figure(fig, 'fair_inheritance_sweden_tax', subfolder='applications') else: # Placeholder figure fig, ax = setup_figure(width=8, height=5) ax.text(0.5, 0.5, 'OECD inheritance tax revenue data\nnot available', ha='center', va='center', fontsize=14, transform=ax.transAxes) ax.set_title('Tax Revenue Confirmation — SKIPPED') paths += save_figure(fig, 'fair_inheritance_sweden_tax', subfolder='applications') # --- Figure 7: Norway DID + parallel trends --- nor_panel = _cache.get('norway_panel') if nor_panel is not None and len(nor_panel) > 0: fig, axes = setup_subplots(1, 2, width=14, height=6) ax = axes[0] for cc in ['NOR', 'DNK', 'FIN']: sub = nor_panel[nor_panel['country_code'] == cc].sort_values('year') if len(sub) > 0: ax.plot(sub['year'], sub['outcome'], 'o-', color=colors.get(cc, 'gray'), label=cc, markersize=4, linewidth=1.5) ax.axvline(2014, color='black', linestyle='--', alpha=0.7, label='Abolition (2014)') ax.set_xlabel('Year') ax.set_ylabel('Top 1% Wealth Share') ax.set_title('Norway 2014: Country Time Series') ax.legend(fontsize=9) ax = axes[1] for label, name, color in [(1, 'Norway (treated)', '#2ecc71'), (0, 'Controls (DK, FI)', '#3498db')]: grp = nor_panel[nor_panel['treated'] == label].groupby('year')['outcome'].mean() grp = grp.sort_index() ax.plot(grp.index, grp.values, 'o-', color=color, label=name, markersize=4, linewidth=1.5) ax.axvline(2014, color='black', linestyle='--', alpha=0.7, label='Abolition (2014)') ax.set_xlabel('Year') ax.set_ylabel('Top 1% Wealth Share') ax.set_title('Norway 2014 DID: Treatment vs Control') ax.legend(fontsize=9) fig.suptitle('Norway 2014 Inheritance Tax Abolition — DID', fontsize=13, y=1.02) paths += save_figure(fig, 'fair_inheritance_norway_did', subfolder='applications') return paths if __name__ == '__main__': results = main() for r in results: print(f"{r['name']}: {r['verdict']} " f"(stat={r['statistic']}, p={r['p_value']})")

Sweden Tax Revenue Confirmation

Description

Figures

Statistics

Script

▸ Source Code

Reproduction Steps