jochen-hanisch/visible-learning
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Erweiterung: Signifikanz-Visualisierung ergänzt und Export sowie Themometer aktualisiert

Browse Source
This commit is contained in:
Jochen Hanisch-Johannsen
2025-09-03 17:15:03 +02:00
parent ff33a5a296
commit 6d8e50e2a3
9 changed files with 681 additions and 444 deletions
Download Patch File Download Diff File Expand all files Collapse all files

161
visible-learning.py
View File

@ -18,7 +18,6 @@ import json
from sklearn.preprocessing import OneHotEncoder
from sklearn.cluster import KMeans
from sklearn.metrics import silhouette_score
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.decomposition import PCA
from sklearn.metrics import silhouette_score, silhouette_samples
@ -101,6 +100,8 @@ def load_data(csv_path: str) -> pd.DataFrame:
df["Effektstärke"].astype(str).str.replace(",", ".", regex=False).str.strip()
)
df["Effektstärke"] = pd.to_numeric(df["Effektstärke"], errors="coerce")
# explizit ±inf auf NaN setzen, um sie zu entfernen
df["Effektstärke"] = df["Effektstärke"].replace([np.inf, -np.inf], np.nan)
# Kapitel aus Thermometer_ID ableiten und Kapitelname mappen
df["Kapitel"] = df["Thermometer_ID"].astype(str).str.split(".").str[0].astype(int)
@ -171,7 +172,7 @@ def add_manual_bins(df: pd.DataFrame) -> pd.DataFrame:
# K-Means-Clustering (Effektstärke + Kapitel)
# -----------------------------------------
def encode_features(df: pd.DataFrame) -> tuple[np.ndarray, list[str]]:
def encode_features(df: pd.DataFrame, kapitel_weight: float = 1.0) -> tuple[np.ndarray, list[str]]:
"""One-Hot-Encoding des Kapitels + Effektstärke (metrisch)."""
try:
enc = OneHotEncoder(sparse_output=False, handle_unknown="ignore") # neuere sklearn-Versionen
@ -179,13 +180,14 @@ def encode_features(df: pd.DataFrame) -> tuple[np.ndarray, list[str]]:
enc = OneHotEncoder(sparse=False, handle_unknown="ignore") # ältere sklearn-Versionen
cat = df[["Kapitel"]].fillna(-1)
cat_ohe = enc.fit_transform(cat)
cat_ohe = cat_ohe * float(kapitel_weight)
eff = df[["Effektstärke"]].values
X = np.hstack([eff, cat_ohe])
feature_names = ["Effektstärke"] + [f"kap::{c}" for c in enc.get_feature_names_out(["Kapitel"])]
return X, feature_names
def encode_features_3d(df: pd.DataFrame) -> tuple[np.ndarray, list[str]]:
def encode_features_3d(df: pd.DataFrame, kapitel_weight: float = 1.0) -> tuple[np.ndarray, list[str]]:
"""Effektstärke + Kapitel + Textdimension (TF-IDF + PCA) für 3D-Clustering."""
# Kapitel
try:
@ -194,6 +196,7 @@ def encode_features_3d(df: pd.DataFrame) -> tuple[np.ndarray, list[str]]:
enc = OneHotEncoder(sparse=False, handle_unknown="ignore")
cat = df[["Kapitel"]].fillna(-1)
cat_ohe = enc.fit_transform(cat)
cat_ohe = cat_ohe * float(kapitel_weight)
# Effektstärke
eff = df[["Effektstärke"]].values
@ -201,6 +204,8 @@ def encode_features_3d(df: pd.DataFrame) -> tuple[np.ndarray, list[str]]:
# Textdimension über TF-IDF + PCA
vectorizer = TfidfVectorizer(max_features=100)
X_text = vectorizer.fit_transform(df["Stichwort"].astype(str))
# Sicherstellen, dass TF-IDF keine Inf/NaN enthält (sollte nicht vorkommen)
X_text = X_text.tocsr()
pca = PCA(n_components=1, random_state=42)
text_dim = pca.fit_transform(X_text.toarray())
@ -212,9 +217,26 @@ def encode_features_3d(df: pd.DataFrame) -> tuple[np.ndarray, list[str]]:
feature_names = ["Effektstärke"] + list(enc.get_feature_names_out(["Kapitel"])) + ["Text_Dimension"]
return X, feature_names
# -----------------------------------------
# Hilfsfunktion zur Sanitisierung von Feature-Matrizen
# -----------------------------------------
def _sanitize_X(X: np.ndarray, clip: float | None = None) -> np.ndarray:
"""Ersetzt NaN/Inf in Feature-Matrizen und optionales Clipping gegen numerische Ausreißer.
Gibt eine *neue* Matrix zurück.
"""
X = np.asarray(X, dtype=float).copy()
# NaN/Inf -> 0
X[~np.isfinite(X)] = 0.0
if clip is not None and clip > 0:
X = np.clip(X, -float(clip), float(clip))
return X
def run_kmeans(df: pd.DataFrame, k: int = 4, random_state: int = 42):
X, feature_names = encode_features(df)
def run_kmeans(df: pd.DataFrame, k: int = 4, random_state: int = 42, kapitel_weight: float = 1.0):
X, feature_names = encode_features(df, kapitel_weight=kapitel_weight)
X = _sanitize_X(X, clip=1e6)
if not np.isfinite(X).all():
print("Warnung: Nicht-endliche Werte in X nach Sanitisierung – werden als 0 behandelt.")
model = KMeans(n_clusters=k, n_init=20, random_state=random_state)
labels = model.fit_predict(X)
sil = silhouette_score(X, labels) if k > 1 and len(df) > k else np.nan
@ -362,14 +384,14 @@ def chi2_bins_kapitel(df: pd.DataFrame):
print(f"Chi²={chi2[0]:.3f}, p={chi2[1]:.6f}, df={chi2[2]} (Unabhängigkeitstest)")
return ct
def cluster_diagnostics(df: pd.DataFrame, k_min: int = 2, k_max: int = 8):
X, _ = encode_features(df)
def cluster_diagnostics(df: pd.DataFrame, k_min: int = 2, k_max: int = 8, kapitel_weight: float = 0.0):
X, _ = encode_features(df, kapitel_weight=kapitel_weight)
inertias, sils, ks = [], [], []
for k in range(k_min, k_max+1):
for k in range(k_min, k_max + 1):
km = KMeans(n_clusters=k, n_init=20, random_state=42).fit(X)
inertias.append(km.inertia_)
ks.append(k)
sils.append(silhouette_score(X, km.labels_) if k>1 else np.nan)
sils.append(silhouette_score(X, km.labels_) if k > 1 else np.nan)
colors = plotly_template.get_colors()
fig = go.Figure()
fig.add_trace(go.Scatter(x=ks, y=inertias, mode="lines+markers",
@ -423,6 +445,7 @@ def build_significance_view(df: pd.DataFrame) -> pd.DataFrame:
- score_cluster = Silhouette_point (kleiner 0 -> auf 0 gesetzt), anschließend min-max-normalisiert
- Gesamt-Score = 0.6*norm(|d|) + 0.4*norm(max(Silhouette_point, 0))
Vorzeichen des Scores folgt dem Vorzeichen von d, damit negative Effekte unten landen.
Hinweis: Clustering/Score in dieser Ansicht wird kapitelunabhängig berechnet, indem Kapitel-OHE mit Gewicht 0.0 skaliert wird.
"""
tmp = df.copy()
# Basisgrößen
@ -505,9 +528,9 @@ def plot_significance_space(df_sig: pd.DataFrame):
))
# Referenzlinien
fig.add_hline(y=0, line=dict(color=colors["border"], width=1))
fig.add_hline(y=0, line=dict(color=colors.get("depthArea"), width=1))
for x0 in [0.0, 0.40, 0.70, -0.40, -0.70]:
fig.add_vline(x=x0, line=dict(color=colors["border"], width=1, dash="dot"))
fig.add_vline(x=x0, line=dict(color=colors.get("depthArea"), width=1, dash="dot"))
fig.update_layout(plotly_template.get_standard_layout(
"Signifikanz-geführter Raum: Effektstärke × Score (kapitelunabhängig)",
@ -543,7 +566,7 @@ def plot_heatmap_kapitel_vs_d(df: pd.DataFrame, kapitel: int | None = None, bins
scale.append([float(t), f"rgb({r},{g},{b})"])
return scale
colorscale = _two_color_scale(colors["depthArea"], colors["brightArea"]) if "depthArea" in colors else "Viridis"
colorscale = _two_color_scale(colors.get("depthArea", "#444"), colors.get("brightArea", "#fff")) if "depthArea" in colors and "brightArea" in colors else colors.get("continuous", "Viridis")
# Histogram2d
fig = go.Figure(data=go.Histogram2d(
@ -684,51 +707,111 @@ def plot_bins(df: pd.DataFrame, kapitel: int | None = None):
export_figure(fig, "vl-bins", export_fig_visual, export_fig_png)
def plot_scatter(df: pd.DataFrame, cluster_labels: np.ndarray, model: KMeans, sil: float, title_suffix: str, kapitel: int | None = None):
def plot_scatter(df: pd.DataFrame, cluster_labels: np.ndarray, model: KMeans, sil: float, title_suffix: str, kapitel: int | None = None, top_n: int = 5):
"""
Kapitelunabhängiger 2D-Scatter:
- x: künstlicher Index, aber so angeordnet, dass Punkte je Cluster zusammenstehen
- y: Effektstärke (Cohen d)
- Farben: Cluster
Zusätzlich:
• horizontale Linien bei den Cluster-Mitteln (Ø d)
• Labels für die Top-N nach |d|
"""
styles = plotly_template.get_plot_styles()
colors = plotly_template.get_colors()
kapitel_label = f"Kapitel {kapitel}" if kapitel else "Gesamt"
tmp = df.copy()
tmp["Cluster"] = cluster_labels.astype(int)
# Plot-X: Kapitel als ganze Zahlen; kleine Jitter-Verschiebung, damit Punkte nicht exakt übereinander liegen
rng = np.random.default_rng(42)
tmp["_kapitel_x"] = tmp["Kapitel"].astype(int) + (rng.random(len(tmp)) - 0.5) * 0.12
# Clusterstärken (Mittelwert der Effektstärke im jeweiligen Clusterzentrum)
cluster_strengths = {i: float(model.cluster_centers_[i][0]) for i in range(len(model.cluster_centers_))}
tmp["Clusterstärke"] = tmp["Cluster"].map(cluster_strengths)
# Cluster-Reihenfolge: absteigend nach Ø d
clusters_sorted = sorted(tmp["Cluster"].unique(), key=lambda c: cluster_strengths[c], reverse=True)
# Gewünschte Markerpalette (robust mit Fallbacks)
def _get_marker(*candidates):
for key in candidates:
if key in styles:
return styles[key]
return styles.get("marker_accent", {})
palette_markers = [
_get_marker("marker_positiveHighlight", "marker_brightArea", "marker_accent"),
_get_marker("marker_primaryLine", "marker_brightArea", "marker_accent"),
_get_marker("marker_secondaryLine", "marker_accent", "marker_brightArea"),
_get_marker("marker_negativeHighlight", "marker_accent", "marker_brightArea"),
]
# x-Positionen so vergeben, dass Cluster-Blöcke entstehen
tmp = tmp.reset_index(drop=True)
tmp["_x"] = np.nan
x_cursor = 0
block_bounds = {} # für Centroid-Linien (x-Min/x-Max je Cluster)
for c in clusters_sorted:
sub_idx = tmp.index[tmp["Cluster"] == c].tolist()
n = len(sub_idx)
xs = np.arange(x_cursor, x_cursor + n)
tmp.loc[sub_idx, "_x"] = xs
block_bounds[c] = (xs.min(), xs.max())
x_cursor += n + 2 # +2 als optischer Abstand zwischen Blöcken
hovertemplate = (
"Thermometer: %{customdata[2]}<br>"
"Stichwort: %{text}<br>"
"Effektstärke: %{y:.2f}<br>"
"Kapitel: %{customdata[0]}<br>"
"Clusterstärke: %{customdata[1]:.3f}<extra></extra>"
"Clusterstärke: %{customdata[1]:.2f}<extra></extra>"
)
fig = go.Figure()
clusters = sorted(tmp["Cluster"].unique())
palette_keys = ["positiveHighlight", "negativeHighlight", "accent", "brightArea"]
for idx, cluster in enumerate(clusters):
cluster_df = tmp[tmp["Cluster"] == cluster]
color_key = palette_keys[idx % len(palette_keys)]
# Punkte je Cluster zeichnen
for idx, c in enumerate(clusters_sorted):
cdf = tmp[tmp["Cluster"] == c]
fig.add_trace(go.Scatter(
x=cluster_df["_kapitel_x"],
y=cluster_df["Effektstärke"],
x=cdf["_x"],
y=cdf["Effektstärke"],
mode="markers",
marker={**styles[f"marker_{color_key}"], "size": 10},
name=f"Cluster: {cluster_strengths[cluster]:.2f}",
text=cluster_df["Stichwort"],
customdata=np.stack([cluster_df["Kapitelname"], cluster_df["Clusterstärke"], cluster_df["Thermometer_ID"]], axis=-1),
marker={**palette_markers[idx % len(palette_markers)], "size": 10},
name=f"Cluster: {cluster_strengths[c]:.2f}",
text=cdf["Stichwort"],
customdata=np.stack([cdf["Kapitelname"], cdf["Clusterstärke"], cdf["Thermometer_ID"]], axis=-1),
hovertemplate=hovertemplate
))
# Centroid-Linien (horizontale Ø d pro Cluster)
for c in clusters_sorted:
x0, x1 = block_bounds[c]
yd = cluster_strengths[c]
centroid_color = colors.get("depthArea", "#444")
line_style = dict(styles.get("linie_secondaryLine", {"width": 2}))
line_style["color"] = centroid_color
fig.add_trace(go.Scatter(
x=[x0, x1],
y=[yd, yd],
mode="lines",
line=line_style,
name=None,
showlegend=False,
hovertemplate=f"Cluster-Mittel: {yd:.2f}<extra></extra>"
))
# Vertikale Trennlinien zwischen Cluster-Blöcken (zur Orientierung)
# (nur als dezente Linien, keine Legende)
block_edges = sorted({bounds[1] + 1 for bounds in block_bounds.values()})
for edge in block_edges[:-1]: # letzte Kante führt bereits zum Abstand
fig.add_vline(x=edge - 1, line=dict(color=colors.get("depthArea"), width=1, dash="dot"))
fig.update_layout(plotly_template.get_standard_layout(
f"Effektstärke × Cluster ({title_suffix}) ({kapitel_label}) – Silhouette: {sil:.3f}", "Kapitel", "Cohen d"
f"Effektstärke × Cluster ({title_suffix}) ({kapitel_label}) – Silhouette: {sil:.3f}",
"Thermometer (gruppiert nach Cluster)", "Cohen d"
))
# Ganze Zahlen auf der x‑Achse (Kapitel)
fig.update_layout(xaxis=dict(tickmode="linear", dtick=1))
fig.update_xaxes(showticklabels=False)
fig.show()
export_figure(fig, f"vl-scatter-{title_suffix}", export_fig_visual, export_fig_png)
@ -757,12 +840,13 @@ def plot_scatter_3d(df: pd.DataFrame, cluster_labels: np.ndarray, sil: float, ti
for idx, cluster in enumerate(clusters):
cluster_df = tmp[tmp["Cluster"] == cluster]
color_key = palette_keys[idx % len(palette_keys)]
marker_style = styles.get(f"marker_{color_key}", {})
fig.add_trace(go.Scatter3d(
x=cluster_df["Effektstärke"],
y=cluster_df["Kapitel"],
z=cluster_df["Text_Dimension"],
mode="markers",
marker={**styles[f"marker_{color_key}"], "size": 6},
marker={**marker_style, "size": 6},
name=f"Cluster {cluster} (Ø d = {cluster_strengths[cluster]:.2f})",
text=cluster_df["Stichwort"],
customdata=np.stack([cluster_df["Kapitelname"], cluster_df["Cluster"]], axis=-1),
@ -807,10 +891,12 @@ def analyse(csv_path: str = "Thermometer.csv", k: int = 4, kapitel: int | None =
df = add_manual_bins(df)
# K-Means
labels, sil, model = run_kmeans(df, k=k)
# Kapitelgewicht = 0.0 => Kapitel-OHE trägt nicht zur Distanz bei (kapitelübergreifendes Clustering)
labels, sil, model = run_kmeans(df, k=k, kapitel_weight=0.0)
# Silhouette je Punkt anhängen
try:
X_for_sil, _ = encode_features(df)
X_for_sil, _ = encode_features(df, kapitel_weight=0.0)
X_for_sil = _sanitize_X(X_for_sil, clip=1e6)
if k > 1 and len(df) > k:
df["Silhouette_point"] = silhouette_samples(X_for_sil, labels)
else:
@ -852,7 +938,7 @@ def analyse(csv_path: str = "Thermometer.csv", k: int = 4, kapitel: int | None =
text_vs_effect(df)
if kapitel is None:
chi2_bins_kapitel(df)
cluster_diagnostics(df)
cluster_diagnostics(df, kapitel_weight=0.0)
profiles_df = cluster_profiles(df, labels)
try:
export_json(json.loads(profiles_df.to_json(orient="table")), "cluster_profile.json")
@ -966,7 +1052,8 @@ def analyse(csv_path: str = "Thermometer.csv", k: int = 4, kapitel: int | None =
plot_scatter(df, labels, model, sil, title_suffix=f"k{k}", kapitel=kapitel)
# 3D-Clustering
X3d, _ = encode_features_3d(df)
X3d, _ = encode_features_3d(df, kapitel_weight=0.0)
X3d = _sanitize_X(X3d, clip=1e6)
model3d = KMeans(n_clusters=k, n_init=20, random_state=42)
labels3d = model3d.fit_predict(X3d)
sil3d = silhouette_score(X3d, labels3d) if k > 1 and len(df) > k else np.nan