chore: simplify Overton scripts, update README, add stemwijzer.db to gitignore
- Extracted EXTREMITY_BUCKET_ORDER constant and _extremity_bucket() helper (4 duplications removed) - Merged two-pass query loop in compute_yearly_baseline into single pass - Removed unused import (mticker), dead code (year_titles_map), 12 obvious comments - Extracted _fmt_axis() helper in SVD drift script - Updated README analysis/ description to include right-wing motion analysis
This commit is contained in:
@@ -30,12 +30,6 @@ import numpy as np
|
||||
|
||||
matplotlib.use("Agg")
|
||||
import matplotlib.pyplot as plt
|
||||
import matplotlib.ticker as mticker
|
||||
|
||||
ROOT = Path(__file__).parent.parent.parent.resolve()
|
||||
if str(ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
from analysis.config import CANONICAL_LEFT, CANONICAL_RIGHT, PARTY_COLOURS
|
||||
|
||||
CANONICAL_CENTRIST = frozenset({"VVD", "D66", "CDA", "NSC", "BBB", "CU"})
|
||||
@@ -47,7 +41,20 @@ DB_PATH = str(ROOT / "data" / "motions.db")
|
||||
REPORTS_DIR = ROOT / "reports" / "overton_window"
|
||||
REPORTS_DIR.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
CANONICAL_CENTRIST_SET = set(CANONICAL_CENTRIST) # nb: config defines as frozenset
|
||||
CANONICAL_CENTRIST_SET = set(CANONICAL_CENTRIST)
|
||||
|
||||
EXTREMITY_BUCKET_ORDER = ["1-2 (mild)", "2-3 (moderate)", "3-4 (high)", "4-5 (extreme)"]
|
||||
|
||||
def _extremity_bucket(score: float) -> str:
|
||||
if score < 2:
|
||||
return "1-2 (mild)"
|
||||
elif score < 3:
|
||||
return "2-3 (moderate)"
|
||||
elif score < 4:
|
||||
return "3-4 (high)"
|
||||
else:
|
||||
return "4-5 (extreme)"
|
||||
|
||||
CANONICAL_LEFT_SET = set(CANONICAL_LEFT)
|
||||
CANONICAL_RIGHT_SET = set(CANONICAL_RIGHT)
|
||||
|
||||
@@ -172,6 +179,7 @@ def compute_yearly_baseline(con: duckdb.DuckDBPyConnection) -> dict[int, dict]:
|
||||
""").fetchall()
|
||||
|
||||
motion_party_votes: dict[int, dict[str, dict[str, int]]] = {}
|
||||
motion_year_map: dict[int, int] = {}
|
||||
for mid, year, party, n, vote in centrist_rows:
|
||||
year = int(year)
|
||||
if year < YEAR_MIN or year > YEAR_MAX:
|
||||
@@ -179,12 +187,7 @@ def compute_yearly_baseline(con: duckdb.DuckDBPyConnection) -> dict[int, dict]:
|
||||
mv = motion_party_votes.setdefault(mid, {})
|
||||
pv = mv.setdefault(party, {"voor": 0, "tegen": 0, "afwezig": 0})
|
||||
pv[vote] = pv.get(vote, 0) + n
|
||||
|
||||
motion_year_map: dict[int, int] = {}
|
||||
for mid, year, _, _, _ in centrist_rows:
|
||||
year = int(year)
|
||||
if YEAR_MIN <= year <= YEAR_MAX:
|
||||
motion_year_map[mid] = year
|
||||
motion_year_map[mid] = year
|
||||
|
||||
for mid, votes in motion_party_votes.items():
|
||||
year = motion_year_map.get(mid)
|
||||
@@ -297,10 +300,6 @@ def compute_opposition_metrics(
|
||||
|
||||
coalition = COALITION
|
||||
|
||||
year_titles_map: dict[int, list[int]] = {}
|
||||
for year, d in yearly_raw.items():
|
||||
year_titles_map[year] = list(range(len(d["titles"])))
|
||||
|
||||
for year, d in yearly_raw.items():
|
||||
coal = coalition.get(year, set())
|
||||
for idx in range(len(d["titles"])):
|
||||
@@ -348,16 +347,9 @@ def compute_extremity_stratified(
|
||||
yearly_raw: dict[int, dict],
|
||||
) -> dict[str, dict[str, list]]:
|
||||
"""Compute centrist_support per extremity bucket, pre vs post 2024."""
|
||||
buckets = {
|
||||
"1-2 (mild)": [],
|
||||
"2-3 (moderate)": [],
|
||||
"3-4 (high)": [],
|
||||
"4-5 (extreme)": [],
|
||||
}
|
||||
|
||||
pre_post: dict[str, dict[str, list]] = {
|
||||
"pre-2024": {b: [] for b in buckets},
|
||||
"post-2024": {b: [] for b in buckets},
|
||||
"pre-2024": {b: [] for b in EXTREMITY_BUCKET_ORDER},
|
||||
"post-2024": {b: [] for b in EXTREMITY_BUCKET_ORDER},
|
||||
}
|
||||
|
||||
for year, d in yearly_raw.items():
|
||||
@@ -367,15 +359,7 @@ def compute_extremity_stratified(
|
||||
cs = d["centrist_support_strict"][idx]
|
||||
if np.isnan(ext) or cs is None or (isinstance(cs, float) and np.isnan(cs)):
|
||||
continue
|
||||
if ext < 2:
|
||||
b = "1-2 (mild)"
|
||||
elif ext < 3:
|
||||
b = "2-3 (moderate)"
|
||||
elif ext < 4:
|
||||
b = "3-4 (high)"
|
||||
else:
|
||||
b = "4-5 (extreme)"
|
||||
pre_post[period][b].append(cs)
|
||||
pre_post[period][_extremity_bucket(ext)].append(cs)
|
||||
|
||||
return pre_post
|
||||
|
||||
@@ -413,12 +397,7 @@ def yearly_summary(yearly: dict[int, dict]) -> dict[int, dict]:
|
||||
|
||||
def sample_audit(yearly_raw: dict[int, dict]) -> list[dict]:
|
||||
"""Stratified random sample: 5 motions per extremity bucket, 20 total."""
|
||||
bucket_motions: dict[str, list[int]] = {
|
||||
"1-2 (mild)": [],
|
||||
"2-3 (moderate)": [],
|
||||
"3-4 (high)": [],
|
||||
"4-5 (extreme)": [],
|
||||
}
|
||||
bucket_motions: dict[str, list[int]] = {b: [] for b in EXTREMITY_BUCKET_ORDER}
|
||||
|
||||
all_motions: list[dict] = []
|
||||
for year, d in yearly_raw.items():
|
||||
@@ -426,14 +405,7 @@ def sample_audit(yearly_raw: dict[int, dict]) -> list[dict]:
|
||||
ext = d["extremity"][idx]
|
||||
if np.isnan(ext):
|
||||
continue
|
||||
if ext < 2:
|
||||
b = "1-2 (mild)"
|
||||
elif ext < 3:
|
||||
b = "2-3 (moderate)"
|
||||
elif ext < 4:
|
||||
b = "3-4 (high)"
|
||||
else:
|
||||
b = "4-5 (extreme)"
|
||||
b = _extremity_bucket(ext)
|
||||
bucket_motions[b].append(len(all_motions))
|
||||
all_motions.append({
|
||||
"year": year,
|
||||
@@ -565,7 +537,6 @@ def create_figure_2(
|
||||
|
||||
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(14, 6))
|
||||
|
||||
# Panel C: Mean extremity over time
|
||||
ax1.plot(years_arr, _vals(yearly_sum, "mean_extremity"),
|
||||
marker="o", color=colour_rw, linewidth=2, label="All right-wing", zorder=5)
|
||||
ax1.plot(years_arr, _vals(opp_sum, "mean_extremity"),
|
||||
@@ -587,8 +558,7 @@ def create_figure_2(
|
||||
ax1.set_xticks(years_arr)
|
||||
ax1.set_xticklabels([str(y) for y in years], rotation=45)
|
||||
|
||||
# Panel D: Extremity-stratified centrist support (grouped bars with IQR error bars)
|
||||
bucket_order = ["1-2 (mild)", "2-3 (moderate)", "3-4 (high)", "4-5 (extreme)"]
|
||||
bucket_order = EXTREMITY_BUCKET_ORDER
|
||||
bucket_labels = ["1-2\nmild", "2-3\nmoderate", "3-4\nhigh", "4-5\nextreme"]
|
||||
bucket_colours = ["#81C784", "#FFB74D", "#E57373", "#BA68C8"]
|
||||
|
||||
@@ -669,7 +639,6 @@ def create_figure_3(
|
||||
means = np.array([left_yearly[y]["mean_left_support"] for y in years])
|
||||
ns = np.array([left_yearly[y]["n"] for y in years])
|
||||
|
||||
# Weighted all-years mean
|
||||
overall_mean = np.average(means, weights=ns) if ns.sum() > 0 else 0.0
|
||||
|
||||
fig, ax = plt.subplots(figsize=(12, 6))
|
||||
@@ -722,11 +691,9 @@ def generate_report(
|
||||
def _val(summary, year, key):
|
||||
return summary[year].get(key, np.nan)
|
||||
|
||||
# Pre/post 2024 comparisons
|
||||
pre_years = [y for y in years if y < BREAK_YEAR]
|
||||
post_years = [y for y in years if y >= BREAK_YEAR]
|
||||
|
||||
# Pooled pre/post values for Cohen's d
|
||||
rw_pre_cs = []
|
||||
rw_post_cs = []
|
||||
rw_pre_ext = []
|
||||
@@ -773,7 +740,6 @@ def generate_report(
|
||||
d_opp_cs = cohens_d(np.array(opp_pre_cs), np.array(opp_post_cs)) if opp_pre_cs and opp_post_cs else float("nan")
|
||||
d_opp_ext = cohens_d(np.array(opp_pre_ext), np.array(opp_post_ext)) if opp_pre_ext and opp_post_ext else float("nan")
|
||||
|
||||
# Yearly summary table
|
||||
yearly_table = "| Year | N (RW) | Centrist Support (Strict) | Extremity | Right Support | Left Opp. |\n"
|
||||
yearly_table += "|------|--------|---------------------------|-----------|---------------|----------|\n"
|
||||
for y in years:
|
||||
@@ -788,8 +754,7 @@ def generate_report(
|
||||
lo_str = f"{lo:.3f}" if not np.isnan(lo) else "N/A"
|
||||
yearly_table += f"| {y} | {int(n)} | {cs_str} | {ext_str} | {rs_str} | {lo_str} |\n"
|
||||
|
||||
# Extremity-stratified table (centrist support)
|
||||
bucket_order = ["1-2 (mild)", "2-3 (moderate)", "3-4 (high)", "4-5 (extreme)"]
|
||||
bucket_order = EXTREMITY_BUCKET_ORDER
|
||||
ext_table = "| Bucket | Period | N | Mean CS | Median CS | P25 | P75 |\n"
|
||||
ext_table += "|--------|--------|---|---------|-----------|---|-----|\n"
|
||||
for b in bucket_order:
|
||||
@@ -815,7 +780,6 @@ def generate_report(
|
||||
f"{pt_p25:.3f} | {pt_p75:.3f} |\n"
|
||||
)
|
||||
|
||||
# Audit table
|
||||
audit_table = "| # | Year | Category | LLM Score | Bucket | Agreed? | Driver |\n"
|
||||
audit_table += "|---|------|----------|-----------|--------|---------|--------|\n"
|
||||
for i, m in enumerate(audit_sample, 1):
|
||||
@@ -901,7 +865,6 @@ def generate_report(
|
||||
"parties filed milder motions post-2024 and the 'shift' is illusory.",
|
||||
]
|
||||
|
||||
# Section 6: Left support for right-wing motions
|
||||
left_years_sorted = sorted(left_yearly.keys())
|
||||
left_pre_years_list = [y for y in pre_years if y in left_yearly]
|
||||
left_post_years_list = [y for y in post_years if y in left_yearly]
|
||||
|
||||
@@ -63,6 +63,10 @@ def _party_in_set(party: str, canonical_set: frozenset) -> bool:
|
||||
return normalized != party and normalized in canonical_set
|
||||
|
||||
|
||||
def _fmt_axis(val: float | None) -> str:
|
||||
return f"{val:.4f}" if val is not None else "N/A"
|
||||
|
||||
|
||||
def compute_aligned_centers(
|
||||
scores: Dict[str, List[List[float]]],
|
||||
windows: List[str],
|
||||
@@ -170,7 +174,6 @@ def compute_drift_metrics(
|
||||
angle_rad = np.arctan2(dy_net, dx_net)
|
||||
angle_deg = float(np.degrees(angle_rad))
|
||||
|
||||
# Right-wing net displacement for comparison
|
||||
right_net = None
|
||||
right_valid = [
|
||||
c for c in annual_centers if c["right_mean_axis1"] is not None
|
||||
@@ -186,7 +189,6 @@ def compute_drift_metrics(
|
||||
"net_dy": round(r_dy, 6),
|
||||
}
|
||||
|
||||
# Is centrist center drifting toward or away from right-wing center?
|
||||
approach_to_right = None
|
||||
if (
|
||||
first.get("right_mean_axis1") is not None
|
||||
@@ -262,7 +264,6 @@ def plot_trajectory(
|
||||
plt.close(fig)
|
||||
return
|
||||
|
||||
# Arrows between consecutive years
|
||||
for i in range(len(cent_a1_valid) - 1):
|
||||
ax.annotate(
|
||||
"",
|
||||
@@ -374,26 +375,10 @@ def write_report(
|
||||
)
|
||||
lines.append("|---|---|---|---|---|---|---|")
|
||||
for c in centers:
|
||||
cent_a1 = (
|
||||
f"{c['centrist_mean_axis1']:.4f}"
|
||||
if c["centrist_mean_axis1"] is not None
|
||||
else "N/A"
|
||||
)
|
||||
cent_a2 = (
|
||||
f"{c['centrist_mean_axis2']:.4f}"
|
||||
if c["centrist_mean_axis2"] is not None
|
||||
else "N/A"
|
||||
)
|
||||
right_a1 = (
|
||||
f"{c['right_mean_axis1']:.4f}"
|
||||
if c["right_mean_axis1"] is not None
|
||||
else "N/A"
|
||||
)
|
||||
right_a2 = (
|
||||
f"{c['right_mean_axis2']:.4f}"
|
||||
if c["right_mean_axis2"] is not None
|
||||
else "N/A"
|
||||
)
|
||||
cent_a1 = _fmt_axis(c["centrist_mean_axis1"])
|
||||
cent_a2 = _fmt_axis(c["centrist_mean_axis2"])
|
||||
right_a1 = _fmt_axis(c["right_mean_axis1"])
|
||||
right_a2 = _fmt_axis(c["right_mean_axis2"])
|
||||
cent_parties = ", ".join(c["centrist_parties_present"])
|
||||
right_parties = ", ".join(c["right_parties_present"])
|
||||
lines.append(
|
||||
@@ -403,7 +388,6 @@ def write_report(
|
||||
|
||||
lines.append("")
|
||||
|
||||
# Drift metrics
|
||||
lines.append("## Drift Metrics (Annual Windows Only)\n")
|
||||
|
||||
if drift.get("net_displacement") is not None:
|
||||
|
||||
Reference in New Issue
Block a user