<jupyter_start><jupyter_text>Generate a classification for the parties where they stay within the political spectrum based on Wahlomat data approach:- append the answer vectors of all elections looked at and cluster- plausibility check > decide for "final model" to classify based on<jupyter_code># load files

import pickle
file = open("data/election_data_since_2010.obj","rb")
election_df = pickle.load(file)
file.close()

file = open("data/twitter_sample.obj","rb")
twitter_df = pickle.load(file)
file.close()
election_df.head(3)
twitter_df.head(3)
# the data needs to be download first from the git hub repro:
# https://github.com/gockelhahn/qual-o-mat-data
## go through all relevant folders (year and import elections within)

import json
import os

years = ["2011","2012","2013","2014","2015","2016","2017","2018","2019","2020","2021"]
base_path = "qual-o-mat-data-master/data/"

def paths_for_year(y,base_path):
    elections = os.listdir(f"{base_path}{y}/")
    return [base_path+y+"/"+x+"/" for x in elections]
paths = []

for y in years:
    paths.extend(paths_for_year(y,base_path))
wahlomat_data = {} # key: yearstate

# three jsons are important for us: opinion (answer vectors)
# party: to translate party encoding into name
# answer: tocompare answer encoding between elections(same encoding?)
# overview: metadata

def import_jsons(path):
    with open(path, "r", encoding="utf-8") as f:
        return json.load(f)
    

# load the differen data for each election into a dict where the key is the election name
for p in paths:
    opinion = import_jsons(f"{p}opinion.json")
    party = import_jsons(f"{p}party.json")
    answer = import_jsons(f"{p}answer.json")
    overview = import_jsons(f"{p}overview.json")
    title = overview["title"]
    date = overview["date"] #yyy-mm-dd
    
    wahlomat_data[f"{title} {date[:4]}"] = {"overview":overview,
                                            "opinion":opinion,
                                            "answer":answer,
                                            "party":party}
    

def party_dict_gen(party_json):
    '''
    input: jsons for one election
    out: dict to translate from party id to party name
    '''
    
    out = {}
    
    for i in party_json:
        out[i["id"]] = i["name"]
        
    return out

def opinion_dict(opinion):
    '''
    input: opinion jason for election
    out: dict containing party id as key and answer vector as item
    '''
    o_dict = {} #each parties digit code will be key and the value a list of its answers to the statements

    for o in opinion:
        party_ = o["party"]
        if party_ in o_dict.keys():
            o_dict[party_].append(o["answer"])
        else:
            o_dict[party_] = [o["answer"]]
    return o_dict
def encode_party(opinion_dict, party_dict):
    '''
    changes key of opinion dict from id to party name
    '''
    new_opinion_dict= {}
    
    for k,v in opinion_dict.items():
        new_opinion_dict[party_dict[k]] = v
    return new_opinion_dict
    
rows = []

# go through each election in the wahlomat dict
for election,item in wahlomat_data.items():
    n = item["overview"]["title"] # get election title
    d = item["overview"]["date"] # get election date
    o = opinion_dict(item["opinion"]) # get opinion dict for election where key=party id and item=answer vector
    p_dict = party_dict_gen(item["party"]) #id to party dict for respective election
    
    
    ##adjust opinion dict party names
    o = encode_party(o,p_dict)
    
    # append read out data to rows list
    rows.append({"election":election,"name":n,"date":d,"opinions":o})
    
import pandas as pd

# generate dataframe from loaded data
wahlomat_df = pd.DataFrame(rows)

### do the clustering for all parties that participated in state and europe elections since 2011
### use the outcome to classify the party sample, republikaner get classified manually/ based on a subclustering based on its elections

# 1. filter wahlomat_ds for relevant elections

cluster_df = wahlomat_df[(wahlomat_df["name"] == "Bundestagswahl") | (wahlomat_df["name"] == "Europawahl")]

# 2. adjust union/cdu csu
## one election they are seperated, add to the other 4 one init for each of them (with same values)

cluster_df.loc[9,"opinions"]["CDU"] = cluster_df.loc[9,"opinions"]['CDU / CSU']
cluster_df.loc[9,"opinions"]["CSU"] = cluster_df.loc[9,"opinions"]['CDU / CSU']

cluster_df.loc[21,"opinions"]["CDU"] = cluster_df.loc[21,"opinions"]['CDU/CSU']
cluster_df.loc[21,"opinions"]["CSU"] = cluster_df.loc[21,"opinions"]['CDU/CSU']

cluster_df.loc[29,"opinions"]["CDU"] = cluster_df.loc[29,"opinions"]['CDU / CSU']
cluster_df.loc[29,"opinions"]["CSU"] = cluster_df.loc[29,"opinions"]['CDU / CSU']

cluster_df.loc[35,"opinions"]["CSU"] = cluster_df.loc[35,"opinions"]['CDU / CSU']
cluster_df.loc[35,"opinions"]["CDU"] = cluster_df.loc[35,"opinions"]['CDU / CSU']

# delete csu/cdu entries
cluster_df.loc[9,"opinions"].pop("CDU / CSU")
cluster_df.loc[21,"opinions"].pop("CDU/CSU")
cluster_df.loc[29,"opinions"].pop("CDU / CSU")
cluster_df.loc[35,"opinions"].pop("CDU / CSU")

cluster_df.head(3)

# 3. check names here and align to filter for parties participating in all 5 elections
parties_wahlomat = []

for op in cluster_df.opinions:
    parties = op.keys()
    parties_wahlomat.extend(parties)
    
print(set(parties_wahlomat))
# to correct

cluster_df.reset_index(drop=True,inplace=True)
cluster_df.loc[0,"opinions"]["BÜNDNIS21"] = cluster_df.loc[0,"opinions"].pop('Bündnis 21/RRP')

cluster_df
parties_wahlomat = []

for op in cluster_df.opinions:
    parties = op.keys()
    parties_wahlomat.extend(parties)
    
print(set(parties_wahlomat))
# chech for parties that run for all 5 election

parties_sets = []

for i in cluster_df["opinions"]:
    
    # add sets for each election to list
    parties_sets.append(set(i.keys()))
    
# get intersection between these sets
print(set.intersection(*parties_sets))
election_df.party.unique()<jupyter_output><empty_output><jupyter_text>comments- Pirates, die partei, ödp and bp participated in all 5 elections- for republicans a seperate clustering will be conducted<jupyter_code>cluster_sample_parties = list(set.intersection(*parties_sets))
opinion_total = dict.fromkeys(cluster_sample_parties, [])

# concatenate the answer vectors for the 5 election for each party except Republikaner
for key in cluster_sample_parties:
    new_vec = []
    for cell in cluster_df["opinions"]:
        new_vec.extend(cell[key])
    opinion_total[key] = new_vec<jupyter_output><empty_output><jupyter_text>Try two different clusterings for robustness- KModes- Agglomerative CLustering  custering Kmodes<jupyter_code>import numpy as np
from kmodes.kmodes import KModes

data = np.array(list(opinion_total.values()))
data
km = KModes(n_clusters=2, init='Huang', n_init=10, verbose=1)

clusters = km.fit_predict(data)
# get clustering
cluster = {}

for i,e in enumerate(km.labels_):
    if e not in cluster.keys():
        cluster[e] = [party[i]["id"]]
    else:
        cluster[e].append(party[i]["id"])
        
cluster = dict(sorted(cluster.items()))
cluster
# translate dict for ids from clustering
cl_p_dict = dict.fromkeys(range(0,15))
for i,p in enumerate(opinion_total.keys()):
    cl_p_dict[i] = p
cl_p_dict
# print cluster
for c,ps in cluster.items():
    print(f"Cluster: {c}")
    for p in ps:
        print(cl_p_dict[p])
# bi-polar clustering left/right leaning fitting common conception<jupyter_output><empty_output><jupyter_text>hirarchical clustering<jupyter_code>import sklearn
from sklearn.cluster import AgglomerativeClustering as AG
clustering = AG().fit(data)
clustering.labels_
wahlomat_data["Bundestagswahl 2013"]["answer"]
### what is needed:

# transform data to ordinal logic:
# switch 1 and 2
## > stimm zu > neutral > stimme nicht zu

data[data==1] = 5 # placeholder

data[data==2] = 1

data[data==5] = 2
data
#https://scikit-learn.org/stable/auto_examples/cluster/plot_agglomerative_dendrogram.html

import numpy as np

from matplotlib import pyplot as plt
from scipy.cluster.hierarchy import dendrogram
from sklearn.datasets import load_iris
from sklearn.cluster import AgglomerativeClustering


def plot_dendrogram(model, **kwargs):
    # Create linkage matrix and then plot the dendrogram

    # create the counts of samples under each node
    counts = np.zeros(model.children_.shape[0])
    n_samples = len(model.labels_)
    for i, merge in enumerate(model.children_):
        current_count = 0
        for child_idx in merge:
            if child_idx < n_samples:
                current_count += 1  # leaf node
            else:
                current_count += counts[child_idx - n_samples]
        counts[i] = current_count

    linkage_matrix = np.column_stack(
        [model.children_, model.distances_, counts]
    ).astype(float)

    # Plot the corresponding dendrogram
    dendrogram(linkage_matrix, **kwargs, labels=list(cl_p_dict.values()))


X = data

# setting distance_threshold=0 ensures we compute the full tree.
model = AgglomerativeClustering(distance_threshold=0, n_clusters=None)

model = model.fit(X)


plt.figure(figsize=(15,3))
plt.title("Hierarchical Clustering Dendrogram without REP")
# plot the top three levels of the dendrogram
plot_dendrogram(model, truncate_mode="level", p=14)
plt.xticks(rotation=45)
plt.savefig("plots/dendo_classification.png",bbox_inches='tight')

plt.show()
# generate dummy variables for political_leaning (0=right,1=left)

clustering.labels_
opinion_total.keys()
# Die PARTEI > DIE 
leaning_dict = {k.upper():v for k,v in zip(opinion_total.keys(),clustering.labels_)}
leaning_dict["LINKE"] = leaning_dict.pop("DIE LINKE")
leaning_dict["REP"] = 0 #manually added do proof !!!!
election_df["leaning"] = [leaning_dict[party] for party in election_df["party"]]
leaning_dict
election_df.head()
import pickle
file = open("data/election_leaning.obj","wb")
pickle.dump(election_df,file)
file.close()<jupyter_output><empty_output><jupyter_text>Proof for classification of Republikaner<jupyter_code># check in which subset REP participated

for i,d in enumerate(cluster_df["opinions"]):
    if "REP" in d.keys():
        print(i)
# first two rows, repeat clustering with that subset

cluster_df_rep = cluster_df.iloc[:2,:]
cluster_df_rep
parties_sets = []

for i in cluster_df_rep["opinions"]:
    parties_sets.append(set(i.keys()))
    
cluster_sample_parties = list(set.intersection(*parties_sets))
opinion_total = dict.fromkeys(cluster_sample_parties, [])

for key in cluster_sample_parties:
    print(key)
    new_vec = []
    for cell in cluster_df_rep["opinions"]:
        new_vec.extend(cell[key])
    opinion_total[key] = new_vec
data = np.array(list(opinion_total.values()))

data[data==1] = 5 # placeholder

data[data==2] = 1

data[data==5] = 2
len(clustering.labels_)
cl_p_dict = dict.fromkeys(range(0,21))
for i,p in enumerate(opinion_total.keys()):
    cl_p_dict[i] = p
#https://scikit-learn.org/stable/auto_examples/cluster/plot_agglomerative_dendrogram.html

import numpy as np

from matplotlib import pyplot as plt
from scipy.cluster.hierarchy import dendrogram
from sklearn.datasets import load_iris
from sklearn.cluster import AgglomerativeClustering


def plot_dendrogram(model, **kwargs):
    # Create linkage matrix and then plot the dendrogram

    # create the counts of samples under each node
    counts = np.zeros(model.children_.shape[0])
    n_samples = len(model.labels_)
    for i, merge in enumerate(model.children_):
        current_count = 0
        for child_idx in merge:
            if child_idx < n_samples:
                current_count += 1  # leaf node
            else:
                current_count += counts[child_idx - n_samples]
        counts[i] = current_count

    linkage_matrix = np.column_stack(
        [model.children_, model.distances_, counts]
    ).astype(float)

    # Plot the corresponding dendrogram
    dendrogram(linkage_matrix, **kwargs, labels=list(cl_p_dict.values()))


X = data

# setting distance_threshold=0 ensures we compute the full tree.
model = AgglomerativeClustering(distance_threshold=0, n_clusters=None)

model = model.fit(X)


plt.figure(figsize=(15,3))
plt.title("Hierarchical Clustering Dendrogram, REP Elections")
# plot the top three levels of the dendrogram
plot_dendrogram(model, truncate_mode="level", p=14)
plt.xlabel("Number of points in node (or index of point if no parenthesis).")
plt.xticks(rotation=45)
plt.savefig("plots/dendo_classification_rep.png",bbox_inches='tight')
plt.show()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11978004462862607, 'fraction_numerical': 0.012910423971947721, 'mean_word_length': 4.281565656565657, 'pattern_counts': {'":': 8, '<': 23, '<?xml version=': 0, '>': 26, 'https://': 3, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 7, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '401918', 'license': 'MIT', 'n_tokens_mistral': 4363, 'n_tokens_neox': 4130, 'n_words': 1050, 'path': '3b_analysis_cluster_parties_wahlomat.ipynb', 'repo_name': 'liowe13088/partyfunding_germany', 'size': '12548'}