#!/usr/bin/env julia using CSV, DataFrames, Random, Dates, SHA const SEED = 20260812 const HOLDOUT_SHARE = 0.20 function region_for(country::AbstractString) europe = Set(["AL","AT","BA","BE","BG","BY","CH","CY","CZ","DE","DK","EE","ES","FI","FR","GB","GE","GR","HR","HU","IE","IS","IT","LT","LU","LV","MD","ME","MK","MT","NL","NO","PL","PT","RO","RS","RU","SE","SI","SK","TR","UA"]) latin = Set(["AR","BO","BR","CL","CO","CR","DO","EC","MX","PA","PE","UY"]) north_america = Set(["CA", "US"]) asia_pacific = Set(["AU", "JP", "KR", "LK", "NZ"]) country in europe && return "Europe" country in latin && return "Latin America" country in north_america && return "North America" country in asia_pacific && return "Asia-Pacific" return "Other" end function sha256_file(path::String) open(path) do io return bytes2hex(sha256(io)) end end function main(args=ARGS) repo_root = normpath(joinpath(@__DIR__, "..")) output_dir = isempty(args) ? joinpath(repo_root, "_local", "revision", "blocked_party") : abspath(args[1]) mkpath(output_dir) mkpath(joinpath(output_dir, "data")) text_path = joinpath(repo_root, "data", "text_data.csv") expert_path = joinpath(repo_root, "data", "expert.csv") lr_path = joinpath(repo_root, "data", "lr_data.csv") union_path = joinpath(repo_root, "data", "union_mapping.csv") text = CSV.read(text_path, DataFrame) expert = CSV.read(expert_path, DataFrame) lr = CSV.read(lr_path, DataFrame) unions = CSV.read(union_path, DataFrame) union_ids = Set(vcat(Int.(unions.manifesto_pf_id), Int.(unions.expert_pf_id))) protected_ids = union(union_ids, Set([1375])) # preserve CDU identification anchor and avoid union leakage text_parties = Set(Int.(unique(text.party))) expert_parties = Set(Int.(unique(expert.party))) candidate_ids = sort(collect(intersect(text_parties, expert_parties))) candidate_rows = NamedTuple[] for party in candidate_ids party in protected_ids && continue t = text[text.party .== party, :] e = expert[expert.party .== party, :] isempty(e.year) && continue text_years = length(unique(t.year)) text_years < 3 && continue has_economic_text = any((t.type_high .== "pro_market") .| (t.type_high .== "pro_welfare")) has_cultural_text = any((t.type_high .== "cosmopolitan") .| (t.type_high .== "traditional")) (has_economic_text && has_cultural_text) || continue country = string(first(e.country)) first_expert_year = minimum(e.year) period = first_expert_year < 1990 ? "pre-1990" : first_expert_year < 2000 ? "1990s" : first_expert_year < 2010 ? "2000s" : first_expert_year < 2020 ? "2010s" : "2020s" push!(candidate_rows, ( party_id=party, country=country, region=region_for(country), first_expert_year=first_expert_year, last_expert_year=maximum(e.year), expert_period=period, text_years=text_years, expert_rows=nrow(e), lr_rows=count(lr.party .== party) )) end candidates = DataFrame(candidate_rows) candidates.stratum = candidates.region .* " / " .* candidates.expert_period rng = MersenneTwister(SEED) selected_ids = Int[] for stratum in sort(unique(candidates.stratum)) ids = sort(candidates.party_id[candidates.stratum .== stratum]) shuffle!(rng, ids) n_select = max(1, round(Int, HOLDOUT_SHARE * length(ids))) n_select = min(n_select, length(ids)) append!(selected_ids, ids[1:n_select]) end sort!(unique!(selected_ids)) selected = candidates[in.(candidates.party_id, Ref(Set(selected_ids))), :] selected.selected = trues(nrow(selected)) expert_test = expert[in.(Int.(expert.party), Ref(Set(selected_ids))), :] expert_train = expert[.!in.(Int.(expert.party), Ref(Set(selected_ids))), :] lr_test = lr[in.(Int.(lr.party), Ref(Set(selected_ids))), :] lr_train = lr[.!in.(Int.(lr.party), Ref(Set(selected_ids))), :] # All text remains in training, which ensures that the held-out parties are # estimated from their text trajectories and the model's shared structure. CSV.write(joinpath(output_dir, "text_data.csv"), text) CSV.write(joinpath(output_dir, "expert.csv"), expert_train) CSV.write(joinpath(output_dir, "lr_data.csv"), lr_train) CSV.write(joinpath(output_dir, "expert_test.csv"), expert_test) CSV.write(joinpath(output_dir, "lr_data_test.csv"), lr_test) CSV.write(joinpath(output_dir, "blocked_parties.csv"), selected) cp(union_path, joinpath(output_dir, "data", "union_mapping.csv"), force=true) # Assertions against leakage and accidental loss of the text evidence. @assert isempty(intersect(Set(Int.(expert_train.party)), Set(selected_ids))) @assert isempty(intersect(Set(Int.(lr_train.party)), Set(selected_ids))) @assert Set(selected_ids) ⊆ Set(Int.(text.party)) @assert nrow(expert_train) + nrow(expert_test) == nrow(expert) @assert nrow(lr_train) + nrow(lr_test) == nrow(lr) manifest = DataFrame( field = [ "created_at", "seed", "holdout_share", "eligible_parties", "blocked_parties", "text_rows_train", "expert_rows_train", "expert_rows_test", "lr_rows_train", "lr_rows_test", "text_sha256", "expert_full_sha256", "lr_full_sha256", "union_mapping_sha256" ], value = [ string(now()), string(SEED), string(HOLDOUT_SHARE), string(nrow(candidates)), string(length(selected_ids)), string(nrow(text)), string(nrow(expert_train)), string(nrow(expert_test)), string(nrow(lr_train)), string(nrow(lr_test)), sha256_file(text_path), sha256_file(expert_path), sha256_file(lr_path), sha256_file(union_path) ] ) CSV.write(joinpath(output_dir, "blocked_validation_manifest.csv"), manifest) stratum_summary = combine(groupby(candidates, :stratum), nrow => :eligible_parties, :party_id => (x -> count(in(Set(selected_ids)), x)) => :blocked_parties) CSV.write(joinpath(output_dir, "blocked_validation_strata.csv"), stratum_summary) println("Prepared blocked validation at: $output_dir") println("Eligible parties: $(nrow(candidates))") println("Blocked parties: $(length(selected_ids))") println("Held-out dimension-specific expert rows: $(nrow(expert_test))") println("Held-out general left-right rows: $(nrow(lr_test))") println("No selected party remains in either expert training file.") end main()