party2d/validation/prepare_blocked_validation.jl

#!/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()