Restore missing core source files (RIPPER variants + experiment runner)

These six modules were referenced throughout the repo (and by the README's
`python run_all_experiments.py` entry point) but had never been committed, so the
code could not run. Recovered from the original working repo and verified byte-for-byte
against the leftover __pycache__ bytecode to be the exact versions last run here.

- run_all_experiments.py (Excel-based data loading + experiment orchestration)
- RIPPER_no_onto_guid.py
- RIPPER_onto_guid.py
- RIPPER_onto_guid_grow.py
- RIPPER_onto_guid_grow_and_prune.py
- RIPPER_onto_guid_grow_prune_mdl.py

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Beau 2026-06-25 17:35:57 +02:00
parent c817872191
commit 742959d674
6 changed files with 2808 additions and 0 deletions

437
RIPPER_no_onto_guid.py Normal file
View file

@ -0,0 +1,437 @@
# Binary one-vs-rest version of jrip_my_version_with_ai.py.
# For each class (rarest first), this trains a binary RIPPER: target class vs everything else.
# At prediction time, the first class whose rule fires wins; if no rule fires, fall back to the
# most-common class — same scheme as evaluate_ruleset.py.
from math import log, comb, log2
from pathlib import Path
import pandas as pd
from pandas import DataFrame
EXCEL_DIR = Path(__file__).parent / "excels"
def shuffle_data(df):
df_new = df.sample(frac=1)
return df_new
def load_data(dataset_suffix="less_classes"):
"""Load excels/X_{suffix}.xlsx + y_{suffix}.xlsx and set the module globals
(all_data, sorted_values, rule_set, possible_conditions) that the run
function reads. Returns nothing — mutates module state."""
global all_data, sorted_values, rule_set, possible_conditions
df = pd.read_excel(EXCEL_DIR / f'X_{dataset_suffix}.xlsx', index_col=0)
y_df = pd.read_excel(EXCEL_DIR / f'y_{dataset_suffix}.xlsx')
df['y'] = y_df['y'].values
all_data = shuffle_data(df)
value_counts = all_data["y"].value_counts().reset_index()
value_counts.columns = ["y", "count"]
sorted_values = value_counts.sort_values("count", ascending=True)["y"].tolist()
rule_set = {value: [] for value in sorted_values}
possible_conditions = all_data.columns.drop('y')
return all_data
def safe_log2(x):
return log2(x) if x > 0 else 0
def safe_log(x):
return log(x) if x > 0 else 0
def safe_comb(n, k):
if n < 0 or k < 0 or k > n:
return 1
return comb(n, k)
def apply_cond_mask(df, cond):
"""Return a boolean Series: True where rows satisfy cond.
Negative conditions are represented with a leading '!' (e.g. '!col' means col==False)."""
if cond.startswith('!'):
return df[cond[1:]] == False
return df[cond] == True
def split_grow_prune_binary(data, target_value):
"""Binary 2/3 - 1/3 split, stratified on target vs not-target (matches wittgenstein's one-vs-rest).
Positives and negatives are each shuffled then split, so both grow and prune sets contain
representative numbers from both classes."""
positives = data[data['y'] == target_value].sample(frac=1)
negatives = data[data['y'] != target_value].sample(frac=1)
pos_split = int(len(positives) * (2/3))
neg_split = int(len(negatives) * (2/3))
if len(positives) <= 2:
grow = pd.concat([positives, negatives.iloc[:neg_split]])
prune = pd.concat([positives, negatives.iloc[neg_split:]])
else:
grow = pd.concat([positives.iloc[:pos_split], negatives.iloc[:neg_split]])
prune = pd.concat([positives.iloc[pos_split:], negatives.iloc[neg_split:]])
return shuffle_data(grow), shuffle_data(prune)
def grow_and_prune_rule(old_P, old_T, P, T, grow_set_new, prune_set, alternative_bool = False, old_rule = None, value = None):
index = 0
rule = list(old_rule) if old_rule else []
P = grow_set_new[grow_set_new['y'] == value].shape[0]
T = grow_set_new.shape[0]
old_P = 0
old_T = 0
while not_perfect(old_P, old_T, P, T, index):
old_P = P
old_T = T
P, T, grow_set_new = grow(value, grow_set_new, rule, P, T)
index += 1
candidates = [rule[:i] for i in range(len(rule), -1, -1)]
prune_value, best_pruned_rule, final_P, final_T = prune(candidates, prune_set, value, alternative_bool)
return best_pruned_rule, final_P, final_T
def not_perfect(old_P, old_T, P, T, index):
best_gain = 0
if P != 0 and T != 0 and old_P != 0 and old_T != 0:
best_gain = P*(safe_log(P/T)-safe_log(old_P/old_T))
if P != 0 and T != 0 and P == T:
return False # perfect, stop
if best_gain < 0 or (best_gain == 0 and index > 0):
return False
return True
def grow(correct_y, all_x, rule, P, T):
max_info_gain_tuples = (0, '', pd.DataFrame())
final_p = 0
final_t = 0
for col in all_x.columns.drop('y'):
for val in [True, False]:
cond_name = ('!' + col) if not val else col
if cond_name in rule: # skip conditions already in the rule
continue
candidate_rule_outcome = all_x[apply_cond_mask(all_x, cond_name)]
t = candidate_rule_outcome.shape[0]
new_data = candidate_rule_outcome[candidate_rule_outcome['y'] == correct_y]
p = new_data.shape[0]
if p == 0 or t == 0:
information_gain = 0
else:
information_gain = p * (safe_log(p/t) - safe_log(P/T))
# WEKA tie-breaking: when two conditions have the same info_gain, prefer the
# MORE SPECIFIC one (smaller `t` = covers fewer rows). With strict `>` the first
# column iterated wins ties — that's an artifact of column order in the dataframe,
# not a meaningful choice. Tie-breaking on smaller `t` produces tighter rules and
# makes grow's output deterministic w.r.t. column order.
better = information_gain > max_info_gain_tuples[0]
tie_break = (information_gain == max_info_gain_tuples[0]
and information_gain > 0
and t < final_t)
if better or tie_break:
final_new_data = new_data.copy(deep=True)
final_p = p
final_t = t
max_info_gain_tuples = (information_gain, cond_name, final_new_data)
condition = max_info_gain_tuples[1]
final_new_data = max_info_gain_tuples[2]
if not condition:
return P, T, all_x
rule.append(condition)
final_new_data = all_x[apply_cond_mask(all_x, condition)]
return final_p, final_t, final_new_data
def prune(candidates, prune_set, value, alternative_bool):
""" Incrementally prune each rule and allow the pruning of any final sequences of the antecedents and the pruning metric is: (p+1)/(p+n+2)"""
highest_candidate_rule_value = (0, [], 0, 1)
all_positives = prune_set[prune_set['y'] == value].shape[0]
all_negatives = prune_set[prune_set['y'] != value].shape[0]
if all_positives + all_negatives == 0:
return highest_candidate_rule_value
for candidate_rule in candidates:
prune_set_filtered = prune_set
if alternative_bool:
for antecedent in candidate_rule:
prune_set_filtered = prune_set_filtered.loc[apply_cond_mask(prune_set_filtered, antecedent)]
not_covered = prune_set.loc[~prune_set.index.isin(prune_set_filtered.index)]
true_negatives = not_covered.loc[not_covered['y'] != value].shape[0]
true_positives = prune_set_filtered.loc[prune_set_filtered['y'] == value].shape[0]
all_positives = prune_set.loc[prune_set['y'] == value].shape[0]
all_negatives = prune_set.loc[prune_set['y'] != value].shape[0]
candidate_rule_value = (true_positives + true_negatives) / (all_positives + all_negatives)
if candidate_rule_value > highest_candidate_rule_value[0]:
highest_candidate_rule_value = (candidate_rule_value, candidate_rule, true_positives, true_negatives+true_positives)
else:
for antecedent in candidate_rule:
prune_set_filtered = prune_set_filtered.loc[apply_cond_mask(prune_set_filtered, antecedent)]
n = prune_set_filtered.loc[prune_set_filtered['y'] != value].shape[0]
p = prune_set_filtered.loc[prune_set_filtered['y'] == value].shape[0]
candidate_rule_value = (p+1)/(p+n+2)
if candidate_rule_value > highest_candidate_rule_value[0]:
highest_candidate_rule_value = (candidate_rule_value, candidate_rule, p, n+p)
return highest_candidate_rule_value
def compute_binary_dl(rules, value, data, possible_conditions):
"""Binary DL: cost to encode rules for ONE class (target=value) + exceptions vs everything else.
Unlike the multi-class compute_ruleset_dl, this does NOT iterate over other classes — each
one-vs-rest sub-problem has its own independent DL, like wittgenstein."""
total_dl = 0
n = 2 * len(possible_conditions) # each column has both a True and False condition
# 1. Theory DL: cost of the rules themselves
for rule in rules:
if not rule:
continue
k = len(rule)
# Cohen 1995 (paper): log2(n+1) fixed-length encoding for k. Cost of "which k of n" is
# split into log2(n+1) for k itself + the combinatorial term for choosing which k conditions.
# This is stricter than wittgenstein's safe_log2(k) — it makes short rules more expensive
# and biases the build/delete stages toward keeping fewer rules.
theory_dl = (k * safe_log2(n/k) + (n-k) * safe_log2(n/(n-k)) + safe_log2(n+1)) * 0.5
total_dl += theory_dl
# 2. Coverage: rows where ANY rule for this class fires
covered = pd.Series(False, index=data.index)
for rule in rules:
if not rule:
continue
mask = pd.Series(True, index=data.index)
for cond in rule:
mask &= apply_cond_mask(data, cond)
covered |= mask
# 3. Exception DL: FP bits (covered rows that aren't target) + FN bits (target rows not covered)
fp_count = data[covered & (data['y'] != value)].shape[0]
total_covered = data[covered].shape[0]
fn_count = data[~covered & (data['y'] == value)].shape[0]
total_not_covered = data[~covered].shape[0]
fp_bits = safe_log2(safe_comb(total_covered, fp_count))
fn_bits = safe_log2(safe_comb(total_not_covered, fn_count))
total_dl += (fp_bits + fn_bits)
return total_dl
def stop_conditions(current_dl, best_dl):
if current_dl > best_dl + 64:
return True
return False
def grow_and_prune_rule(old_P, old_T, P, T, grow_set_new, prune_set, alternative_bool = False, old_rule = None, value = None):
index = 0
rule = list(old_rule) if old_rule else []
P = grow_set_new[grow_set_new['y'] == value].shape[0]
T = grow_set_new.shape[0]
old_P = 0
old_T = 0
while not_perfect(old_P, old_T, P, T, index):
old_P = P
old_T = T
P, T, grow_set_new = grow(value, grow_set_new, rule, P, T)
index += 1
candidates = [rule[:i] for i in range(len(rule), -1, -1)]
prune_value, best_pruned_rule, final_P, final_T = prune(candidates, prune_set, value, alternative_bool)
return best_pruned_rule, final_P, final_T
def print_and_save_to_file(rule_set):
with open('final_ruleset.txt', 'w') as f:
for key, value in rule_set.items():
f.write(f"label: {key} rule: {value}\n")
def get_covered_mask(df, rule):
mask = pd.Series(True, index=df.index)
for cond in rule:
mask &= apply_cond_mask(df, cond)
return mask
def run_RIPPER_without_guidance(train_df=None):
"""One-vs-rest binary RIPPER, like wittgenstein. For each class, train an independent binary
classifier (target vs all others). Each binary sub-problem gets its own grow/prune split with
ALL non-target rows as negatives, so the prune set is stable even for rare classes.
If train_df (bool feature cols + a 'y' column) is given, train on it; otherwise fall back to
the module globals set by load_data()/the caller. train_df is NOT shuffled here — callers seed
numpy's global RNG and the per-class grow/prune splits handle randomisation.
Returns (rule_set, sorted_values)."""
global all_data, sorted_values, rule_set, possible_conditions
if train_df is not None:
all_data = train_df
value_counts = all_data["y"].value_counts().reset_index()
value_counts.columns = ["y", "count"]
sorted_values = value_counts.sort_values("count", ascending=True)["y"].tolist()
rule_set = {value: [] for value in sorted_values}
possible_conditions = all_data.columns.drop("y")
# MIN_PRECISION: at prediction time we still do multi-class voting, so the relevant baseline is
# the default class frequency in the full multi-class problem (not the 99% "majority not-target"
# of the binary sub-problem, which would block everything).
MIN_PRECISION = (all_data['y'] == sorted_values[-1]).sum() / len(all_data)
for value in sorted_values[:-1]:
# Per-class binary splits: target class vs all others. Three independent shuffles for Stage 2.
grow_set_local, prune_set_local = split_grow_prune_binary(all_data, value)
grow_set_replace_local, prune_set_replace_local = split_grow_prune_binary(all_data, value)
grow_set_revise_local, prune_set_revise_local = split_grow_prune_binary(all_data, value)
build_index = 1
# Initial DL: ruleset with no rules for this class (target rows are all FN, no FPs)
min_total_dl = compute_binary_dl([], value, all_data, possible_conditions)
curr_grow = grow_set_local.copy()
curr_prune = prune_set_local.copy()
# --- STAGE 1: BUILD ---
while True:
P = curr_grow[curr_grow['y'] == value].shape[0]
T = curr_grow.shape[0]
if P == 0:
break
rule, _, _ = grow_and_prune_rule(0, 0, P, T, curr_grow, curr_prune, alternative_bool=False, value=value)
if not rule:
break
# Reject rules whose precision on the FULL dataset is below the default class frequency.
full_mask = get_covered_mask(all_data, rule)
full_T = int(full_mask.sum())
full_P = int((all_data[full_mask]['y'] == value).sum())
if full_T > 0 and full_P / full_T < MIN_PRECISION:
break
new_dl = compute_binary_dl(rule_set[value] + [rule], value, all_data, possible_conditions)
if new_dl < min_total_dl:
min_total_dl = new_dl
if stop_conditions(new_dl, min_total_dl):
break
rule_set[value].append(rule)
grow_mask = get_covered_mask(curr_grow, rule)
prune_mask = get_covered_mask(curr_prune, rule)
curr_grow = curr_grow[~grow_mask]
curr_prune = curr_prune[~prune_mask]
build_index += 1
# --- STAGE 2: OPTIMIZE EACH RULE FOR THAT SPECIFIC RULE SET ---
# WEKA's JRip defaults to m_Optimizations = 2 — it runs the replacement/revision pass
# twice over each class's ruleset, generating fresh stratified splits each pass.
M_OPTIMIZATIONS = 2
for opt_pass in range(M_OPTIMIZATIONS):
# Fresh splits for this optimization pass.
grow_set_replace_local, prune_set_replace_local = split_grow_prune_binary(all_data, value)
grow_set_revise_local, prune_set_revise_local = split_grow_prune_binary(all_data, value)
for index in range(len(rule_set[value])):
current_rule = rule_set[value][index]
dl_orig = compute_binary_dl(rule_set[value], value, all_data, possible_conditions)
# Variant 1: Replacement
grow_set_new_replace = grow_set_replace_local.copy(deep=True)
prune_set_new_replace = prune_set_replace_local.copy(deep=True)
for new_index in range(len(rule_set[value])):
prev_clause = rule_set[value][new_index]
if new_index == index:
break
mask_grow = get_covered_mask(grow_set_new_replace, prev_clause)
mask_prune = get_covered_mask(prune_set_new_replace, prev_clause)
grow_set_new_replace = grow_set_new_replace[~mask_grow]
prune_set_new_replace = prune_set_new_replace[~mask_prune]
P = grow_set_new_replace[grow_set_new_replace['y'] == value].shape[0]
T = grow_set_new_replace.shape[0]
old_P = 0
old_T = 0
# Stage 2 uses accuracy (TP+TN)/(P+N) per the paper.
rule_replace, replace_P, _ = grow_and_prune_rule(old_P, old_T, P, T, grow_set_new_replace, prune_set_new_replace, True, None, value)
dl_alt1 = compute_binary_dl(rule_set[value][:index] + [rule_replace] + rule_set[value][index + 1:], value, all_data, possible_conditions)
# Variant 2: Revision (start from current rule, add more conditions)
grow_set_new_revise = grow_set_revise_local.copy(deep=True)
prune_set_new_revise = prune_set_revise_local.copy(deep=True)
for new_index in range(len(rule_set[value])):
if new_index == index:
break
prev_clause = rule_set[value][new_index]
mask_grow = get_covered_mask(grow_set_new_revise, prev_clause)
mask_prune = get_covered_mask(prune_set_new_revise, prev_clause)
grow_set_new_revise = grow_set_new_revise[~mask_grow]
prune_set_new_revise = prune_set_new_revise[~mask_prune]
for antecedent in current_rule:
grow_set_new_revise = grow_set_new_revise[apply_cond_mask(grow_set_new_revise, antecedent)]
prune_set_new_revise = prune_set_new_revise[apply_cond_mask(prune_set_new_revise, antecedent)]
P = grow_set_new_revise[grow_set_new_revise['y'] == value].shape[0]
T = grow_set_new_revise.shape[0]
old_P = 0
old_T = 0
rule_revise, revise_P, _ = grow_and_prune_rule(old_P, old_T, P, T, grow_set_new_revise, prune_set_new_revise, True, list(current_rule), value)
dl_alt2 = compute_binary_dl(rule_set[value][:index] + [rule_revise] + rule_set[value][index + 1:], value, all_data, possible_conditions)
if not rule_replace or replace_P == 0:
dl_alt1 = float('inf')
if not rule_revise or revise_P == 0:
dl_alt2 = float('inf')
if dl_orig <= dl_alt1 and dl_orig <= dl_alt2:
pass
elif dl_alt1 < dl_orig and dl_alt1 <= dl_alt2:
rule_set[value][index] = rule_replace
elif dl_alt2 < dl_orig and dl_alt2 < dl_alt1:
rule_set[value][index] = rule_revise
# --- STAGE 3: COVER REMAINING POSITIVES ---
covered_indices = set()
for rule in rule_set[value]:
mask = get_covered_mask(all_data, rule)
covered_indices.update(all_data[mask].index)
residual_positives = all_data[(all_data['y'] == value) & ~all_data.index.isin(covered_indices)]
if not residual_positives.empty:
# Fresh binary split: residual positives vs all original negatives.
negatives = all_data[all_data['y'] != value]
residual_data = shuffle_data(pd.concat([residual_positives, negatives]))
curr_grow, curr_prune = split_grow_prune_binary(residual_data, value)
build_index = 0
while True:
P = curr_grow[curr_grow['y'] == value].shape[0]
T = curr_grow.shape[0]
if P == 0:
break
rule, _, _ = grow_and_prune_rule(0, 0, P, T, curr_grow, curr_prune, alternative_bool=False, value=value)
if not rule:
break
full_mask = get_covered_mask(all_data, rule)
full_T = int(full_mask.sum())
full_P = int((all_data[full_mask]['y'] == value).sum())
if full_T > 0 and full_P / full_T < MIN_PRECISION:
break
new_dl = compute_binary_dl(rule_set[value] + [rule], value, all_data, possible_conditions)
if new_dl < min_total_dl:
min_total_dl = new_dl
if stop_conditions(new_dl, min_total_dl):
break
rule_set[value].append(rule)
grow_mask = get_covered_mask(curr_grow, rule)
prune_mask = get_covered_mask(curr_prune, rule)
curr_grow = curr_grow[~grow_mask]
curr_prune = curr_prune[~prune_mask]
build_index += 1
# --- STAGE 4: DELETE RULES THAT INCREASE DL (per binary sub-problem) ---
# Each class has its own independent binary DL — delete rules that don't pay for themselves
# against the target-vs-rest objective only.
current_dl = compute_binary_dl(rule_set[value], value, all_data, possible_conditions)
for rule in list(rule_set[value]):
idx = rule_set[value].index(rule)
candidate_rules = rule_set[value][:idx] + rule_set[value][idx+1:]
candidate_dl = compute_binary_dl(candidate_rules, value, all_data, possible_conditions)
# WEKA uses strict `<` here: a rule is only deleted if removing it STRICTLY lowers DL.
# With `<=`, ties would discard rules that pay for themselves exactly — costing recall
# without any DL benefit.
if candidate_dl < current_dl:
current_dl = candidate_dl
rule_set[value] = rule_set[value][:idx] + rule_set[value][idx+1:]
return rule_set, sorted_values

439
RIPPER_onto_guid.py Normal file
View file

@ -0,0 +1,439 @@
# Binary one-vs-rest version of jrip_my_version_with_ai.py.
# For each class (rarest first), this trains a binary RIPPER: target class vs everything else.
# At prediction time, the first class whose rule fires wins; if no rule fires, fall back to the
# most-common class — same scheme as evaluate_ruleset.py.
from math import log, comb, log2
from pathlib import Path
import pandas as pd
from pandas import DataFrame
from ontology_guidance.ontology_guidance import compute_ontology_viol
EXCEL_DIR = Path(__file__).parent / "excels"
def shuffle_data(df):
df_new = df.sample(frac=1)
return df_new
def load_data(dataset_suffix="less_classes"):
"""Load excels/X_{suffix}.xlsx + y_{suffix}.xlsx and set the module globals
(all_data, sorted_values, rule_set, possible_conditions) that the run
function reads. Returns nothing — mutates module state."""
global all_data, sorted_values, rule_set, possible_conditions
df = pd.read_excel(EXCEL_DIR / f'X_{dataset_suffix}.xlsx', index_col=0)
y_df = pd.read_excel(EXCEL_DIR / f'y_{dataset_suffix}.xlsx')
df['y'] = y_df['y'].values
all_data = shuffle_data(df)
value_counts = all_data["y"].value_counts().reset_index()
value_counts.columns = ["y", "count"]
sorted_values = value_counts.sort_values("count", ascending=True)["y"].tolist()
rule_set = {value: [] for value in sorted_values}
possible_conditions = all_data.columns.drop('y')
return all_data
def safe_log2(x):
return log2(x) if x > 0 else 0
def safe_log(x):
return log(x) if x > 0 else 0
def safe_comb(n, k):
if n < 0 or k < 0 or k > n:
return 1
return comb(n, k)
def apply_cond_mask(df, cond):
"""Return a boolean Series: True where rows satisfy cond.
Negative conditions are represented with a leading '!' (e.g. '!col' means col==False)."""
if cond.startswith('!'):
return df[cond[1:]] == False
return df[cond] == True
def split_grow_prune_binary(data, target_value):
"""Binary 2/3 - 1/3 split, stratified on target vs not-target (matches wittgenstein's one-vs-rest).
Positives and negatives are each shuffled then split, so both grow and prune sets contain
representative numbers from both classes."""
positives = data[data['y'] == target_value].sample(frac=1)
negatives = data[data['y'] != target_value].sample(frac=1)
pos_split = int(len(positives) * (2/3))
neg_split = int(len(negatives) * (2/3))
if len(positives) <= 2:
grow = pd.concat([positives, negatives.iloc[:neg_split]])
prune = pd.concat([positives, negatives.iloc[neg_split:]])
else:
grow = pd.concat([positives.iloc[:pos_split], negatives.iloc[:neg_split]])
prune = pd.concat([positives.iloc[pos_split:], negatives.iloc[neg_split:]])
return shuffle_data(grow), shuffle_data(prune)
def grow_and_prune_rule(old_P, old_T, P, T, grow_set_new, prune_set, alternative_bool = False, old_rule = None, value = None):
index = 0
rule = list(old_rule) if old_rule else []
P = grow_set_new[grow_set_new['y'] == value].shape[0]
T = grow_set_new.shape[0]
old_P = 0
old_T = 0
while not_perfect(old_P, old_T, P, T, index):
old_P = P
old_T = T
P, T, grow_set_new = grow(value, grow_set_new, rule, P, T)
index += 1
candidates = [rule[:i] for i in range(len(rule), -1, -1)]
prune_value, best_pruned_rule, final_P, final_T = prune(candidates, prune_set, value, alternative_bool)
return best_pruned_rule, final_P, final_T
def not_perfect(old_P, old_T, P, T, index):
best_gain = 0
if P != 0 and T != 0 and old_P != 0 and old_T != 0:
best_gain = P*(safe_log(P/T)-safe_log(old_P/old_T))
if P != 0 and T != 0 and P == T:
return False # perfect, stop
if best_gain < 0 or (best_gain == 0 and index > 0):
return False
return True
def grow(correct_y, all_x, rule, P, T):
max_info_gain_tuples = (0, '', pd.DataFrame())
final_p = 0
final_t = 0
for col in all_x.columns.drop('y'):
for val in [True, False]:
cond_name = ('!' + col) if not val else col
if cond_name in rule: # skip conditions already in the rule
continue
candidate_rule_outcome = all_x[apply_cond_mask(all_x, cond_name)]
t = candidate_rule_outcome.shape[0]
new_data = candidate_rule_outcome[candidate_rule_outcome['y'] == correct_y]
p = new_data.shape[0]
if p == 0 or t == 0:
information_gain = 0
else:
information_gain = p * (safe_log(p/t) - safe_log(P/T))
# WEKA tie-breaking: when two conditions have the same info_gain, prefer the
# MORE SPECIFIC one (smaller `t` = covers fewer rows). With strict `>` the first
# column iterated wins ties — that's an artifact of column order in the dataframe,
# not a meaningful choice. Tie-breaking on smaller `t` produces tighter rules and
# makes grow's output deterministic w.r.t. column order.
better = information_gain > max_info_gain_tuples[0]
tie_break = (information_gain == max_info_gain_tuples[0]
and information_gain > 0
and t < final_t)
if better or tie_break:
final_new_data = new_data.copy(deep=True)
final_p = p
final_t = t
max_info_gain_tuples = (information_gain, cond_name, final_new_data)
condition = max_info_gain_tuples[1]
final_new_data = max_info_gain_tuples[2]
if not condition:
return P, T, all_x
rule.append(condition)
final_new_data = all_x[apply_cond_mask(all_x, condition)]
return final_p, final_t, final_new_data
def prune(candidates, prune_set, value, alternative_bool):
""" Incrementally prune each rule and allow the pruning of any final sequences of the antecedents and the pruning metric is: (p+1)/(p+n+2)"""
highest_candidate_rule_value = (0, [], 0, 1)
all_positives = prune_set[prune_set['y'] == value].shape[0]
all_negatives = prune_set[prune_set['y'] != value].shape[0]
if all_positives + all_negatives == 0:
return highest_candidate_rule_value
for candidate_rule in candidates:
prune_set_filtered = prune_set
if alternative_bool:
for antecedent in candidate_rule:
prune_set_filtered = prune_set_filtered.loc[apply_cond_mask(prune_set_filtered, antecedent)]
not_covered = prune_set.loc[~prune_set.index.isin(prune_set_filtered.index)]
true_negatives = not_covered.loc[not_covered['y'] != value].shape[0]
true_positives = prune_set_filtered.loc[prune_set_filtered['y'] == value].shape[0]
all_positives = prune_set.loc[prune_set['y'] == value].shape[0]
all_negatives = prune_set.loc[prune_set['y'] != value].shape[0]
candidate_rule_value = (true_positives + true_negatives) / (all_positives + all_negatives)
if candidate_rule_value > highest_candidate_rule_value[0]:
highest_candidate_rule_value = (candidate_rule_value, candidate_rule, true_positives, true_negatives+true_positives)
else:
for antecedent in candidate_rule:
prune_set_filtered = prune_set_filtered.loc[apply_cond_mask(prune_set_filtered, antecedent)]
n = prune_set_filtered.loc[prune_set_filtered['y'] != value].shape[0]
p = prune_set_filtered.loc[prune_set_filtered['y'] == value].shape[0]
candidate_rule_value = (p+1)/(p+n+2)
if candidate_rule_value > highest_candidate_rule_value[0]:
highest_candidate_rule_value = (candidate_rule_value, candidate_rule, p, n+p)
return highest_candidate_rule_value
def compute_binary_dl(rules, value, data, possible_conditions, lambda_val):
"""Binary DL: cost to encode rules for ONE class (target=value) + exceptions vs everything else.
Unlike the multi-class compute_ruleset_dl, this does NOT iterate over other classes — each
one-vs-rest sub-problem has its own independent DL, like wittgenstein."""
total_dl = 0
n = 2 * len(possible_conditions) # each column has both a True and False condition
# 1. Theory DL: cost of the rules themselves
for rule in rules:
if not rule:
continue
k = len(rule)
# Cohen 1995 (paper): log2(n+1) fixed-length encoding for k. Cost of "which k of n" is
# split into log2(n+1) for k itself + the combinatorial term for choosing which k conditions.
# This is stricter than wittgenstein's safe_log2(k) — it makes short rules more expensive
# and biases the build/delete stages toward keeping fewer rules.
theory_dl = (k * safe_log2(n/k) + (n-k) * safe_log2(n/(n-k)) + safe_log2(n+1)) * 0.5
total_dl += theory_dl
# 2. Coverage: rows where ANY rule for this class fires
covered = pd.Series(False, index=data.index)
for rule in rules:
if not rule:
continue
mask = pd.Series(True, index=data.index)
for cond in rule:
mask &= apply_cond_mask(data, cond)
covered |= mask
# 3. Exception DL: FP bits (covered rows that aren't target) + FN bits (target rows not covered)
fp_count = data[covered & (data['y'] != value)].shape[0]
total_covered = data[covered].shape[0]
fn_count = data[~covered & (data['y'] == value)].shape[0]
total_not_covered = data[~covered].shape[0]
fp_bits = safe_log2(safe_comb(total_covered, fp_count))
fn_bits = safe_log2(safe_comb(total_not_covered, fn_count))
total_dl += (fp_bits + fn_bits)
# add ontology violation penalty
ont_violations = compute_ontology_viol(rules, value, data)
total_dl += ont_violations * lambda_val
return total_dl
def stop_conditions(current_dl, best_dl):
if current_dl > best_dl + 64:
return True
return False
def grow_and_prune_rule(old_P, old_T, P, T, grow_set_new, prune_set, alternative_bool = False, old_rule = None, value = None):
index = 0
rule = list(old_rule) if old_rule else []
P = grow_set_new[grow_set_new['y'] == value].shape[0]
T = grow_set_new.shape[0]
old_P = 0
old_T = 0
while not_perfect(old_P, old_T, P, T, index):
old_P = P
old_T = T
P, T, grow_set_new = grow(value, grow_set_new, rule, P, T)
index += 1
candidates = [rule[:i] for i in range(len(rule), -1, -1)]
prune_value, best_pruned_rule, final_P, final_T = prune(candidates, prune_set, value, alternative_bool)
return best_pruned_rule, final_P, final_T
def print_and_save_to_file(rule_set):
with open('final_ruleset_onto_guid.txt', 'w') as f:
for key, value in rule_set.items():
f.write(f"label: {key} rule: {value}\n")
def get_covered_mask(df, rule):
mask = pd.Series(True, index=df.index)
for cond in rule:
mask &= apply_cond_mask(df, cond)
return mask
def run_RIPPER_with_guidance(train_df=None, lambda_val=1):
"""One-vs-rest binary RIPPER, like wittgenstein. For each class, train an independent binary
classifier (target vs all others). Each binary sub-problem gets its own grow/prune split with
ALL non-target rows as negatives, so the prune set is stable even for rare classes.
If train_df (bool feature cols + a 'y' column) is given, train on it; otherwise fall back to
the module globals set by load_data()/the caller. train_df is NOT shuffled here — callers seed
numpy's global RNG and the per-class grow/prune splits handle randomisation.
Returns (rule_set, sorted_values)."""
global all_data, sorted_values, rule_set, possible_conditions
if train_df is not None:
all_data = train_df
value_counts = all_data["y"].value_counts().reset_index()
value_counts.columns = ["y", "count"]
sorted_values = value_counts.sort_values("count", ascending=True)["y"].tolist()
rule_set = {value: [] for value in sorted_values}
possible_conditions = all_data.columns.drop("y")
# MIN_PRECISION: at prediction time we still do multi-class voting, so the relevant baseline is
# the default class frequency in the full multi-class problem (not the 99% "majority not-target"
# of the binary sub-problem, which would block everything).
MIN_PRECISION = (all_data['y'] == sorted_values[-1]).sum() / len(all_data)
for value in sorted_values[:-1]:
# Per-class binary splits: target class vs all others. Three independent shuffles for Stage 2.
grow_set_local, prune_set_local = split_grow_prune_binary(all_data, value)
grow_set_replace_local, prune_set_replace_local = split_grow_prune_binary(all_data, value)
grow_set_revise_local, prune_set_revise_local = split_grow_prune_binary(all_data, value)
build_index = 1
# Initial DL: ruleset with no rules for this class (target rows are all FN, no FPs)
min_total_dl = compute_binary_dl([], value, all_data, possible_conditions, lambda_val=lambda_val)
curr_grow = grow_set_local.copy()
curr_prune = prune_set_local.copy()
# --- STAGE 1: BUILD ---
while True:
P = curr_grow[curr_grow['y'] == value].shape[0]
T = curr_grow.shape[0]
if P == 0:
break
rule, _, _ = grow_and_prune_rule(0, 0, P, T, curr_grow, curr_prune, alternative_bool=False, value=value)
if not rule:
break
# Reject rules whose precision on the FULL dataset is below the default class frequency.
full_mask = get_covered_mask(all_data, rule)
full_T = int(full_mask.sum())
full_P = int((all_data[full_mask]['y'] == value).sum())
if full_T > 0 and full_P / full_T < MIN_PRECISION:
break
new_dl = compute_binary_dl(rule_set[value] + [rule], value, all_data, possible_conditions, lambda_val=lambda_val)
if new_dl < min_total_dl:
min_total_dl = new_dl
if stop_conditions(new_dl, min_total_dl):
break
rule_set[value].append(rule)
grow_mask = get_covered_mask(curr_grow, rule)
prune_mask = get_covered_mask(curr_prune, rule)
curr_grow = curr_grow[~grow_mask]
curr_prune = curr_prune[~prune_mask]
build_index += 1
# --- STAGE 2: OPTIMIZE EACH RULE FOR THAT SPECIFIC RULE SET ---
# WEKA's JRip defaults to m_Optimizations = 2 — it runs the replacement/revision pass
# twice over each class's ruleset, generating fresh stratified splits each pass.
M_OPTIMIZATIONS = 2
for opt_pass in range(M_OPTIMIZATIONS):
# Fresh splits for this optimization pass.
grow_set_replace_local, prune_set_replace_local = split_grow_prune_binary(all_data, value)
grow_set_revise_local, prune_set_revise_local = split_grow_prune_binary(all_data, value)
for index in range(len(rule_set[value])):
current_rule = rule_set[value][index]
dl_orig = compute_binary_dl(rule_set[value], value, all_data, possible_conditions, lambda_val=lambda_val)
# Variant 1: Replacement
grow_set_new_replace = grow_set_replace_local.copy(deep=True)
prune_set_new_replace = prune_set_replace_local.copy(deep=True)
for new_index in range(len(rule_set[value])):
prev_clause = rule_set[value][new_index]
if new_index == index:
break
mask_grow = get_covered_mask(grow_set_new_replace, prev_clause)
mask_prune = get_covered_mask(prune_set_new_replace, prev_clause)
grow_set_new_replace = grow_set_new_replace[~mask_grow]
prune_set_new_replace = prune_set_new_replace[~mask_prune]
P = grow_set_new_replace[grow_set_new_replace['y'] == value].shape[0]
T = grow_set_new_replace.shape[0]
old_P = 0
old_T = 0
# Stage 2 uses accuracy (TP+TN)/(P+N) per the paper.
rule_replace, replace_P, _ = grow_and_prune_rule(old_P, old_T, P, T, grow_set_new_replace, prune_set_new_replace, True, None, value)
dl_alt1 = compute_binary_dl(rule_set[value][:index] + [rule_replace] + rule_set[value][index + 1:], value, all_data, possible_conditions, lambda_val=lambda_val)
# Variant 2: Revision (start from current rule, add more conditions)
grow_set_new_revise = grow_set_revise_local.copy(deep=True)
prune_set_new_revise = prune_set_revise_local.copy(deep=True)
for new_index in range(len(rule_set[value])):
if new_index == index:
break
prev_clause = rule_set[value][new_index]
mask_grow = get_covered_mask(grow_set_new_revise, prev_clause)
mask_prune = get_covered_mask(prune_set_new_revise, prev_clause)
grow_set_new_revise = grow_set_new_revise[~mask_grow]
prune_set_new_revise = prune_set_new_revise[~mask_prune]
for antecedent in current_rule:
grow_set_new_revise = grow_set_new_revise[apply_cond_mask(grow_set_new_revise, antecedent)]
prune_set_new_revise = prune_set_new_revise[apply_cond_mask(prune_set_new_revise, antecedent)]
P = grow_set_new_revise[grow_set_new_revise['y'] == value].shape[0]
T = grow_set_new_revise.shape[0]
old_P = 0
old_T = 0
rule_revise, revise_P, _ = grow_and_prune_rule(old_P, old_T, P, T, grow_set_new_revise, prune_set_new_revise, True, list(current_rule), value)
dl_alt2 = compute_binary_dl(rule_set[value][:index] + [rule_revise] + rule_set[value][index + 1:], value, all_data, possible_conditions, lambda_val=lambda_val)
if not rule_replace or replace_P == 0:
dl_alt1 = float('inf')
if not rule_revise or revise_P == 0:
dl_alt2 = float('inf')
if dl_orig <= dl_alt1 and dl_orig <= dl_alt2:
pass
elif dl_alt1 < dl_orig and dl_alt1 <= dl_alt2:
rule_set[value][index] = rule_replace
elif dl_alt2 < dl_orig and dl_alt2 < dl_alt1:
rule_set[value][index] = rule_revise
# --- STAGE 3: COVER REMAINING POSITIVES ---
covered_indices = set()
for rule in rule_set[value]:
mask = get_covered_mask(all_data, rule)
covered_indices.update(all_data[mask].index)
residual_positives = all_data[(all_data['y'] == value) & ~all_data.index.isin(covered_indices)]
if not residual_positives.empty:
# Fresh binary split: residual positives vs all original negatives.
negatives = all_data[all_data['y'] != value]
residual_data = shuffle_data(pd.concat([residual_positives, negatives]))
curr_grow, curr_prune = split_grow_prune_binary(residual_data, value)
build_index = 0
while True:
P = curr_grow[curr_grow['y'] == value].shape[0]
T = curr_grow.shape[0]
if P == 0:
break
rule, _, _ = grow_and_prune_rule(0, 0, P, T, curr_grow, curr_prune, alternative_bool=False, value=value)
if not rule:
break
full_mask = get_covered_mask(all_data, rule)
full_T = int(full_mask.sum())
full_P = int((all_data[full_mask]['y'] == value).sum())
if full_T > 0 and full_P / full_T < MIN_PRECISION:
break
new_dl = compute_binary_dl(rule_set[value] + [rule], value, all_data, possible_conditions, lambda_val=lambda_val)
if new_dl < min_total_dl:
min_total_dl = new_dl
if stop_conditions(new_dl, min_total_dl):
break
rule_set[value].append(rule)
grow_mask = get_covered_mask(curr_grow, rule)
prune_mask = get_covered_mask(curr_prune, rule)
curr_grow = curr_grow[~grow_mask]
curr_prune = curr_prune[~prune_mask]
build_index += 1
# --- STAGE 4: DELETE RULES THAT INCREASE DL (per binary sub-problem) ---
# Each class has its own independent binary DL — delete rules that don't pay for themselves
# against the target-vs-rest objective only.
current_dl = compute_binary_dl(rule_set[value], value, all_data, possible_conditions, lambda_val=lambda_val)
for rule in list(rule_set[value]):
idx = rule_set[value].index(rule)
candidate_rules = rule_set[value][:idx] + rule_set[value][idx+1:]
candidate_dl = compute_binary_dl(candidate_rules, value, all_data, possible_conditions, lambda_val=lambda_val)
# WEKA uses strict `<` here: a rule is only deleted if removing it STRICTLY lowers DL.
# With `<=`, ties would discard rules that pay for themselves exactly — costing recall
# without any DL benefit.
if candidate_dl < current_dl:
current_dl = candidate_dl
rule_set[value] = rule_set[value][:idx] + rule_set[value][idx+1:]
return rule_set, sorted_values

447
RIPPER_onto_guid_grow.py Normal file
View file

@ -0,0 +1,447 @@
# Binary one-vs-rest version of jrip_my_version_with_ai.py.
# For each class (rarest first), this trains a binary RIPPER: target class vs everything else.
# At prediction time, the first class whose rule fires wins; if no rule fires, fall back to the
# most-common class — same scheme as evaluate_ruleset.py.
from math import log, comb, log2
from pathlib import Path
import pandas as pd
from pandas import DataFrame
from ontology_guidance.ontology_guidance import compute_rule_viol
from helper_functions.build_invalid_row_mask import build_invalid_row_mask
EXCEL_DIR = Path(__file__).parent / "excels"
def shuffle_data(df):
df_new = df.sample(frac=1)
return df_new
def load_data(dataset_suffix="less_classes"):
"""Load excels/X_{suffix}.xlsx + y_{suffix}.xlsx and set the module globals
(all_data, sorted_values, rule_set, possible_conditions) that the run
function reads. Returns nothing — mutates module state."""
global all_data, sorted_values, rule_set, possible_conditions
df = pd.read_excel(EXCEL_DIR / f'X_{dataset_suffix}.xlsx', index_col=0)
y_df = pd.read_excel(EXCEL_DIR / f'y_{dataset_suffix}.xlsx')
df['y'] = y_df['y'].values
all_data = shuffle_data(df)
value_counts = all_data["y"].value_counts().reset_index()
value_counts.columns = ["y", "count"]
sorted_values = value_counts.sort_values("count", ascending=True)["y"].tolist()
rule_set = {value: [] for value in sorted_values}
possible_conditions = all_data.columns.drop('y')
return all_data
def safe_log2(x):
return log2(x) if x > 0 else 0
def safe_log(x):
return log(x) if x > 0 else 0
def safe_comb(n, k):
if n < 0 or k < 0 or k > n:
return 1
return comb(n, k)
def apply_cond_mask(df, cond):
"""Return a boolean Series: True where rows satisfy cond.
Negative conditions are represented with a leading '!' (e.g. '!col' means col==False)."""
if cond.startswith('!'):
return df[cond[1:]] == False
return df[cond] == True
def split_grow_prune_binary(data, target_value):
"""Binary 2/3 - 1/3 split, stratified on target vs not-target (matches wittgenstein's one-vs-rest).
Positives and negatives are each shuffled then split, so both grow and prune sets contain
representative numbers from both classes."""
positives = data[data['y'] == target_value].sample(frac=1)
negatives = data[data['y'] != target_value].sample(frac=1)
pos_split = int(len(positives) * (2/3))
neg_split = int(len(negatives) * (2/3))
if len(positives) <= 2:
grow = pd.concat([positives, negatives.iloc[:neg_split]])
prune = pd.concat([positives, negatives.iloc[neg_split:]])
else:
grow = pd.concat([positives.iloc[:pos_split], negatives.iloc[:neg_split]])
prune = pd.concat([positives.iloc[pos_split:], negatives.iloc[neg_split:]])
return shuffle_data(grow), shuffle_data(prune)
# def grow_and_prune_rule(old_P, old_T, P, T, grow_set_new, prune_set, alternative_bool = False, old_rule = None, value = None):
# index = 0
# rule = list(old_rule) if old_rule else []
# P = grow_set_new[grow_set_new['y'] == value].shape[0]
# T = grow_set_new.shape[0]
# old_P = 0
# old_T = 0
# while not_perfect(old_P, old_T, P, T, index):
# old_P = P
# old_T = T
# P, T, grow_set_new = grow(value, grow_set_new, rule, P, T)
# index += 1
# candidates = [rule[:i] for i in range(len(rule), -1, -1)]
# prune_value, best_pruned_rule, final_P, final_T = prune(candidates, prune_set, value, alternative_bool)
# #print(f"best pruned rule is: {best_pruned_rule} with prune value: {prune_value} and P: {final_P} and T: {final_T}")
# return best_pruned_rule, final_P, final_T
def not_perfect(old_P, old_T, P, T, index):
best_gain = 0
if P != 0 and T != 0 and old_P != 0 and old_T != 0:
best_gain = P*(safe_log(P/T)-safe_log(old_P/old_T))
if P != 0 and T != 0 and P == T:
return False # perfect, stop
if best_gain < 0 or (best_gain == 0 and index > 0):
return False
return True
def grow(correct_y, all_x, rule, P, T, lambda_val, invalid_row_mask):
max_info_gain_tuples = (0, '', pd.DataFrame())
final_p = 0
final_t = 0
for col in all_x.columns.drop('y'):
for val in [True, False]:
cond_name = ('!' + col) if not val else col
if cond_name in rule: # skip conditions already in the rule
continue
candidate_rule_outcome = all_x[apply_cond_mask(all_x, cond_name)]
t = candidate_rule_outcome.shape[0]
new_data = candidate_rule_outcome[candidate_rule_outcome['y'] == correct_y]
p = new_data.shape[0]
if p == 0 or t == 0:
information_gain = 0
else:
information_gain = p * (safe_log(p/t) - safe_log(P/T))
viol = compute_rule_viol(list(rule + [cond_name]), correct_y, all_x, invalid_row_mask)
information_gain = information_gain - lambda_val * viol
# WEKA tie-breaking: when two conditions have the same info_gain, prefer the
# MORE SPECIFIC one (smaller `t` = covers fewer rows). With strict `>` the first
# column iterated wins ties — that's an artifact of column order in the dataframe,
# not a meaningful choice. Tie-breaking on smaller `t` produces tighter rules and
# makes grow's output deterministic w.r.t. column order.
better = information_gain > max_info_gain_tuples[0]
tie_break = (information_gain == max_info_gain_tuples[0]
and information_gain > 0
and t < final_t)
if better or tie_break:
final_new_data = new_data.copy(deep=True)
final_p = p
final_t = t
max_info_gain_tuples = (information_gain, cond_name, final_new_data)
condition = max_info_gain_tuples[1]
final_new_data = max_info_gain_tuples[2]
if not condition:
return P, T, all_x
rule.append(condition)
final_new_data = all_x[apply_cond_mask(all_x, condition)]
return final_p, final_t, final_new_data
def prune(candidates, prune_set, value, alternative_bool):
""" Incrementally prune each rule and allow the pruning of any final sequences of the antecedents and the pruning metric is: (p+1)/(p+n+2)"""
highest_candidate_rule_value = (0, [], 0, 1)
all_positives = prune_set[prune_set['y'] == value].shape[0]
all_negatives = prune_set[prune_set['y'] != value].shape[0]
if all_positives + all_negatives == 0:
return highest_candidate_rule_value
for candidate_rule in candidates:
prune_set_filtered = prune_set
if alternative_bool:
for antecedent in candidate_rule:
prune_set_filtered = prune_set_filtered.loc[apply_cond_mask(prune_set_filtered, antecedent)]
not_covered = prune_set.loc[~prune_set.index.isin(prune_set_filtered.index)]
true_negatives = not_covered.loc[not_covered['y'] != value].shape[0]
true_positives = prune_set_filtered.loc[prune_set_filtered['y'] == value].shape[0]
all_positives = prune_set.loc[prune_set['y'] == value].shape[0]
all_negatives = prune_set.loc[prune_set['y'] != value].shape[0]
candidate_rule_value = (true_positives + true_negatives) / (all_positives + all_negatives)
if candidate_rule_value > highest_candidate_rule_value[0]:
highest_candidate_rule_value = (candidate_rule_value, candidate_rule, true_positives, true_negatives+true_positives)
else:
for antecedent in candidate_rule:
prune_set_filtered = prune_set_filtered.loc[apply_cond_mask(prune_set_filtered, antecedent)]
n = prune_set_filtered.loc[prune_set_filtered['y'] != value].shape[0]
p = prune_set_filtered.loc[prune_set_filtered['y'] == value].shape[0]
candidate_rule_value = (p+1)/(p+n+2)
if candidate_rule_value > highest_candidate_rule_value[0]:
highest_candidate_rule_value = (candidate_rule_value, candidate_rule, p, n+p)
return highest_candidate_rule_value
def compute_binary_dl(rules, value, data, possible_conditions):
"""Binary DL: cost to encode rules for ONE class (target=value) + exceptions vs everything else.
Unlike the multi-class compute_ruleset_dl, this does NOT iterate over other classes — each
one-vs-rest sub-problem has its own independent DL, like wittgenstein."""
total_dl = 0
n = 2 * len(possible_conditions) # each column has both a True and False condition
# 1. Theory DL: cost of the rules themselves
for rule in rules:
if not rule:
continue
k = len(rule)
# Cohen 1995 (paper): log2(n+1) fixed-length encoding for k. Cost of "which k of n" is
# split into log2(n+1) for k itself + the combinatorial term for choosing which k conditions.
# This is stricter than wittgenstein's safe_log2(k) — it makes short rules more expensive
# and biases the build/delete stages toward keeping fewer rules.
theory_dl = (k * safe_log2(n/k) + (n-k) * safe_log2(n/(n-k)) + safe_log2(n+1)) * 0.5
total_dl += theory_dl
# 2. Coverage: rows where ANY rule for this class fires
covered = pd.Series(False, index=data.index)
for rule in rules:
if not rule:
continue
mask = pd.Series(True, index=data.index)
for cond in rule:
mask &= apply_cond_mask(data, cond)
covered |= mask
# 3. Exception DL: FP bits (covered rows that aren't target) + FN bits (target rows not covered)
fp_count = data[covered & (data['y'] != value)].shape[0]
total_covered = data[covered].shape[0]
fn_count = data[~covered & (data['y'] == value)].shape[0]
total_not_covered = data[~covered].shape[0]
fp_bits = safe_log2(safe_comb(total_covered, fp_count))
fn_bits = safe_log2(safe_comb(total_not_covered, fn_count))
total_dl += (fp_bits + fn_bits)
return total_dl
def stop_conditions(current_dl, best_dl):
if current_dl > best_dl + 64:
return True
return False
def grow_and_prune_rule(old_P, old_T, P, T, grow_set_new, prune_set, alternative_bool = False, old_rule = None, value = None, lambda_val=1, invalid_row_mask=None):
index = 0
rule = list(old_rule) if old_rule else []
P = grow_set_new[grow_set_new['y'] == value].shape[0]
T = grow_set_new.shape[0]
old_P = 0
old_T = 0
while not_perfect(old_P, old_T, P, T, index):
old_P = P
old_T = T
P, T, grow_set_new = grow(value, grow_set_new, rule, P, T, lambda_val, invalid_row_mask=invalid_row_mask)
index += 1
candidates = [rule[:i] for i in range(len(rule), -1, -1)]
prune_value, best_pruned_rule, final_P, final_T = prune(candidates, prune_set, value, alternative_bool)
return best_pruned_rule, final_P, final_T
def print_and_save_to_file(rule_set):
with open('final_ruleset.txt', 'w') as f:
for key, value in rule_set.items():
f.write(f"label: {key} rule: {value}\n")
def get_covered_mask(df, rule):
mask = pd.Series(True, index=df.index)
for cond in rule:
mask &= apply_cond_mask(df, cond)
return mask
def run_RIPPER_with_guidance_in_grow(train_df=None, lambda_val=1):
"""One-vs-rest binary RIPPER, like wittgenstein. For each class, train an independent binary
classifier (target vs all others). Each binary sub-problem gets its own grow/prune split with
ALL non-target rows as negatives, so the prune set is stable even for rare classes.
If train_df (bool feature cols + a 'y' column) is given, train on it; otherwise fall back to
the module globals set by load_data()/the caller. train_df is NOT shuffled here — callers seed
numpy's global RNG and the per-class grow/prune splits handle randomisation.
Returns (rule_set, sorted_values)."""
global all_data, sorted_values, rule_set, possible_conditions
if train_df is not None:
all_data = train_df
value_counts = all_data["y"].value_counts().reset_index()
value_counts.columns = ["y", "count"]
sorted_values = value_counts.sort_values("count", ascending=True)["y"].tolist()
rule_set = {value: [] for value in sorted_values}
possible_conditions = all_data.columns.drop("y")
# MIN_PRECISION: at prediction time we still do multi-class voting, so the relevant baseline is
# the default class frequency in the full multi-class problem (not the 99% "majority not-target"
# of the binary sub-problem, which would block everything).
MIN_PRECISION = (all_data['y'] == sorted_values[-1]).sum() / len(all_data)
for value in sorted_values[:-1]:
# Per-class ontology mask, built once on all_data and reused by every grow call
# for this class (Stages 1, 2 replacement/revision, 3). Depends only on (value, all_data).
invalid_row_mask = build_invalid_row_mask(value, all_data)
# Per-class binary splits: target class vs all others. Three independent shuffles for Stage 2.
grow_set_local, prune_set_local = split_grow_prune_binary(all_data, value)
grow_set_replace_local, prune_set_replace_local = split_grow_prune_binary(all_data, value)
grow_set_revise_local, prune_set_revise_local = split_grow_prune_binary(all_data, value)
build_index = 1
# Initial DL: ruleset with no rules for this class (target rows are all FN, no FPs)
min_total_dl = compute_binary_dl([], value, all_data, possible_conditions)
curr_grow = grow_set_local.copy()
curr_prune = prune_set_local.copy()
# --- STAGE 1: BUILD ---
while True:
P = curr_grow[curr_grow['y'] == value].shape[0]
T = curr_grow.shape[0]
if P == 0:
break
rule, _, _ = grow_and_prune_rule(0, 0, P, T, curr_grow, curr_prune, False, None, value, lambda_val, invalid_row_mask)
if not rule:
break
# Reject rules whose precision on the FULL dataset is below the default class frequency.
full_mask = get_covered_mask(all_data, rule)
full_T = int(full_mask.sum())
full_P = int((all_data[full_mask]['y'] == value).sum())
if full_T > 0 and full_P / full_T < MIN_PRECISION:
break
new_dl = compute_binary_dl(rule_set[value] + [rule], value, all_data, possible_conditions)
if new_dl < min_total_dl:
min_total_dl = new_dl
if stop_conditions(new_dl, min_total_dl):
break
rule_set[value].append(rule)
grow_mask = get_covered_mask(curr_grow, rule)
prune_mask = get_covered_mask(curr_prune, rule)
curr_grow = curr_grow[~grow_mask]
curr_prune = curr_prune[~prune_mask]
build_index += 1
# --- STAGE 2: OPTIMIZE EACH RULE FOR THAT SPECIFIC RULE SET ---
# WEKA's JRip defaults to m_Optimizations = 2 — it runs the replacement/revision pass
# twice over each class's ruleset, generating fresh stratified splits each pass.
M_OPTIMIZATIONS = 2
for opt_pass in range(M_OPTIMIZATIONS):
# Fresh splits for this optimization pass.
grow_set_replace_local, prune_set_replace_local = split_grow_prune_binary(all_data, value)
grow_set_revise_local, prune_set_revise_local = split_grow_prune_binary(all_data, value)
for index in range(len(rule_set[value])):
current_rule = rule_set[value][index]
dl_orig = compute_binary_dl(rule_set[value], value, all_data, possible_conditions)
# Variant 1: Replacement
grow_set_new_replace = grow_set_replace_local.copy(deep=True)
prune_set_new_replace = prune_set_replace_local.copy(deep=True)
for new_index in range(len(rule_set[value])):
prev_clause = rule_set[value][new_index]
if new_index == index:
break
mask_grow = get_covered_mask(grow_set_new_replace, prev_clause)
mask_prune = get_covered_mask(prune_set_new_replace, prev_clause)
grow_set_new_replace = grow_set_new_replace[~mask_grow]
prune_set_new_replace = prune_set_new_replace[~mask_prune]
P = grow_set_new_replace[grow_set_new_replace['y'] == value].shape[0]
T = grow_set_new_replace.shape[0]
old_P = 0
old_T = 0
# Stage 2 uses accuracy (TP+TN)/(P+N) per the paper.
rule_replace, replace_P, _ = grow_and_prune_rule(old_P, old_T, P, T, grow_set_new_replace, prune_set_new_replace, True, None, value, lambda_val, invalid_row_mask)
dl_alt1 = compute_binary_dl(rule_set[value][:index] + [rule_replace] + rule_set[value][index + 1:], value, all_data, possible_conditions)
# Variant 2: Revision (start from current rule, add more conditions)
grow_set_new_revise = grow_set_revise_local.copy(deep=True)
prune_set_new_revise = prune_set_revise_local.copy(deep=True)
for new_index in range(len(rule_set[value])):
if new_index == index:
break
prev_clause = rule_set[value][new_index]
mask_grow = get_covered_mask(grow_set_new_revise, prev_clause)
mask_prune = get_covered_mask(prune_set_new_revise, prev_clause)
grow_set_new_revise = grow_set_new_revise[~mask_grow]
prune_set_new_revise = prune_set_new_revise[~mask_prune]
for antecedent in current_rule:
grow_set_new_revise = grow_set_new_revise[apply_cond_mask(grow_set_new_revise, antecedent)]
prune_set_new_revise = prune_set_new_revise[apply_cond_mask(prune_set_new_revise, antecedent)]
P = grow_set_new_revise[grow_set_new_revise['y'] == value].shape[0]
T = grow_set_new_revise.shape[0]
old_P = 0
old_T = 0
rule_revise, revise_P, _ = grow_and_prune_rule(old_P, old_T, P, T, grow_set_new_revise, prune_set_new_revise, True, list(current_rule), value, lambda_val, invalid_row_mask)
dl_alt2 = compute_binary_dl(rule_set[value][:index] + [rule_revise] + rule_set[value][index + 1:], value, all_data, possible_conditions)
if not rule_replace or replace_P == 0:
dl_alt1 = float('inf')
if not rule_revise or revise_P == 0:
dl_alt2 = float('inf')
if dl_orig <= dl_alt1 and dl_orig <= dl_alt2:
pass
elif dl_alt1 < dl_orig and dl_alt1 <= dl_alt2:
rule_set[value][index] = rule_replace
elif dl_alt2 < dl_orig and dl_alt2 < dl_alt1:
rule_set[value][index] = rule_revise
# --- STAGE 3: COVER REMAINING POSITIVES ---
covered_indices = set()
for rule in rule_set[value]:
mask = get_covered_mask(all_data, rule)
covered_indices.update(all_data[mask].index)
residual_positives = all_data[(all_data['y'] == value) & ~all_data.index.isin(covered_indices)]
if not residual_positives.empty:
# Fresh binary split: residual positives vs all original negatives.
negatives = all_data[all_data['y'] != value]
residual_data = shuffle_data(pd.concat([residual_positives, negatives]))
curr_grow, curr_prune = split_grow_prune_binary(residual_data, value)
build_index = 0
while True:
P = curr_grow[curr_grow['y'] == value].shape[0]
T = curr_grow.shape[0]
if P == 0:
break
rule, _, _ = grow_and_prune_rule(0, 0, P, T, curr_grow, curr_prune, False, None, value, lambda_val, invalid_row_mask)
if not rule:
break
full_mask = get_covered_mask(all_data, rule)
full_T = int(full_mask.sum())
full_P = int((all_data[full_mask]['y'] == value).sum())
if full_T > 0 and full_P / full_T < MIN_PRECISION:
break
new_dl = compute_binary_dl(rule_set[value] + [rule], value, all_data, possible_conditions)
if new_dl < min_total_dl:
min_total_dl = new_dl
if stop_conditions(new_dl, min_total_dl):
break
rule_set[value].append(rule)
grow_mask = get_covered_mask(curr_grow, rule)
prune_mask = get_covered_mask(curr_prune, rule)
curr_grow = curr_grow[~grow_mask]
curr_prune = curr_prune[~prune_mask]
build_index += 1
# --- STAGE 4: DELETE RULES THAT INCREASE DL (per binary sub-problem) ---
# Each class has its own independent binary DL — delete rules that don't pay for themselves
# against the target-vs-rest objective only.
current_dl = compute_binary_dl(rule_set[value], value, all_data, possible_conditions)
for rule in list(rule_set[value]):
idx = rule_set[value].index(rule)
candidate_rules = rule_set[value][:idx] + rule_set[value][idx+1:]
candidate_dl = compute_binary_dl(candidate_rules, value, all_data, possible_conditions)
# WEKA uses strict `<` here: a rule is only deleted if removing it STRICTLY lowers DL.
# With `<=`, ties would discard rules that pay for themselves exactly — costing recall
# without any DL benefit.
if candidate_dl < current_dl:
current_dl = candidate_dl
rule_set[value] = rule_set[value][:idx] + rule_set[value][idx+1:]
return rule_set, sorted_values

View file

@ -0,0 +1,432 @@
# Binary one-vs-rest version of jrip_my_version_with_ai.py.
# For each class (rarest first), this trains a binary RIPPER: target class vs everything else.
# At prediction time, the first class whose rule fires wins; if no rule fires, fall back to the
# most-common class — same scheme as evaluate_ruleset.py.
from math import log, comb, log2
from pathlib import Path
import pandas as pd
from pandas import DataFrame
from ontology_guidance.ontology_guidance import compute_rule_viol
from helper_functions.build_invalid_row_mask import build_invalid_row_mask
EXCEL_DIR = Path(__file__).parent / "excels"
def shuffle_data(df):
df_new = df.sample(frac=1)
return df_new
def load_data(dataset_suffix="less_classes"):
"""Load excels/X_{suffix}.xlsx + y_{suffix}.xlsx and set the module globals
(all_data, sorted_values, rule_set, possible_conditions) that the run
function reads. Returns nothing — mutates module state."""
global all_data, sorted_values, rule_set, possible_conditions
df = pd.read_excel(EXCEL_DIR / f'X_{dataset_suffix}.xlsx', index_col=0)
y_df = pd.read_excel(EXCEL_DIR / f'y_{dataset_suffix}.xlsx')
df['y'] = y_df['y'].values
all_data = shuffle_data(df)
value_counts = all_data["y"].value_counts().reset_index()
value_counts.columns = ["y", "count"]
sorted_values = value_counts.sort_values("count", ascending=True)["y"].tolist()
rule_set = {value: [] for value in sorted_values}
possible_conditions = all_data.columns.drop('y')
return all_data
def safe_log2(x):
return log2(x) if x > 0 else 0
def safe_log(x):
return log(x) if x > 0 else 0
def safe_comb(n, k):
if n < 0 or k < 0 or k > n:
return 1
return comb(n, k)
def apply_cond_mask(df, cond):
"""Return a boolean Series: True where rows satisfy cond.
Negative conditions are represented with a leading '!' (e.g. '!col' means col==False)."""
if cond.startswith('!'):
return df[cond[1:]] == False
return df[cond] == True
def split_grow_prune_binary(data, target_value):
"""Binary 2/3 - 1/3 split, stratified on target vs not-target (matches wittgenstein's one-vs-rest).
Positives and negatives are each shuffled then split, so both grow and prune sets contain
representative numbers from both classes."""
positives = data[data['y'] == target_value].sample(frac=1)
negatives = data[data['y'] != target_value].sample(frac=1)
pos_split = int(len(positives) * (2/3))
neg_split = int(len(negatives) * (2/3))
if len(positives) <= 2:
grow = pd.concat([positives, negatives.iloc[:neg_split]])
prune = pd.concat([positives, negatives.iloc[neg_split:]])
else:
grow = pd.concat([positives.iloc[:pos_split], negatives.iloc[:neg_split]])
prune = pd.concat([positives.iloc[pos_split:], negatives.iloc[neg_split:]])
return shuffle_data(grow), shuffle_data(prune)
def not_perfect(old_P, old_T, P, T, index):
best_gain = 0
if P != 0 and T != 0 and old_P != 0 and old_T != 0:
best_gain = P*(safe_log(P/T)-safe_log(old_P/old_T))
if P != 0 and T != 0 and P == T:
return False # perfect, stop
if best_gain < 0 or (best_gain == 0 and index > 0):
return False
return True
def grow(correct_y, all_x, rule, P, T, lambda_val, invalid_row_mask):
max_info_gain_tuples = (0, '', pd.DataFrame())
final_p = 0
final_t = 0
for col in all_x.columns.drop('y'):
for val in [True, False]:
cond_name = ('!' + col) if not val else col
if cond_name in rule: # skip conditions already in the rule
continue
candidate_rule_outcome = all_x[apply_cond_mask(all_x, cond_name)]
t = candidate_rule_outcome.shape[0]
new_data = candidate_rule_outcome[candidate_rule_outcome['y'] == correct_y]
p = new_data.shape[0]
if p == 0 or t == 0:
information_gain = 0
else:
information_gain = p * (safe_log(p/t) - safe_log(P/T))
viol = compute_rule_viol(list(rule + [cond_name]), correct_y, all_x, invalid_row_mask)
information_gain = information_gain - lambda_val * viol
# WEKA tie-breaking: when two conditions have the same info_gain, prefer the
# MORE SPECIFIC one (smaller `t` = covers fewer rows). With strict `>` the first
# column iterated wins ties — that's an artifact of column order in the dataframe,
# not a meaningful choice. Tie-breaking on smaller `t` produces tighter rules and
# makes grow's output deterministic w.r.t. column order.
better = information_gain > max_info_gain_tuples[0]
tie_break = (information_gain == max_info_gain_tuples[0]
and information_gain > 0
and t < final_t)
if better or tie_break:
final_new_data = new_data.copy(deep=True)
final_p = p
final_t = t
max_info_gain_tuples = (information_gain, cond_name, final_new_data)
condition = max_info_gain_tuples[1]
final_new_data = max_info_gain_tuples[2]
if not condition:
return P, T, all_x
rule.append(condition)
final_new_data = all_x[apply_cond_mask(all_x, condition)]
return final_p, final_t, final_new_data
def prune(candidates, prune_set, value, alternative_bool, lambda_val):
""" Incrementally prune each rule and allow the pruning of any final sequences of the antecedents and the pruning metric is: (p+1)/(p+n+2)"""
highest_candidate_rule_value = (0, [], 0, 1)
invalid_row_mask_prune = build_invalid_row_mask(value, prune_set)
all_positives = prune_set[prune_set['y'] == value].shape[0]
all_negatives = prune_set[prune_set['y'] != value].shape[0]
if all_positives + all_negatives == 0:
return highest_candidate_rule_value
for candidate_rule in candidates:
prune_set_filtered = prune_set
n_onto = compute_rule_viol(candidate_rule, value, prune_set, invalid_row_mask_prune)
if alternative_bool:
for antecedent in candidate_rule:
prune_set_filtered = prune_set_filtered.loc[apply_cond_mask(prune_set_filtered, antecedent)]
not_covered = prune_set.loc[~prune_set.index.isin(prune_set_filtered.index)]
true_negatives = not_covered.loc[not_covered['y'] != value].shape[0]
true_positives = prune_set_filtered.loc[prune_set_filtered['y'] == value].shape[0]
all_positives = prune_set.loc[prune_set['y'] == value].shape[0]
all_negatives = prune_set.loc[prune_set['y'] != value].shape[0]
candidate_rule_value = ((true_positives + true_negatives) - (lambda_val * n_onto)) / (all_positives + all_negatives)
if candidate_rule_value > highest_candidate_rule_value[0]:
highest_candidate_rule_value = (candidate_rule_value, candidate_rule, true_positives, true_negatives+true_positives)
else:
for antecedent in candidate_rule:
prune_set_filtered = prune_set_filtered.loc[apply_cond_mask(prune_set_filtered, antecedent)]
n = prune_set_filtered.loc[prune_set_filtered['y'] != value].shape[0]
p = prune_set_filtered.loc[prune_set_filtered['y'] == value].shape[0]
#candidate_rule_value = (p+1)/(p+n+2)
candidate_rule_value = (p + 1) / (p + n + lambda_val * n_onto + 2)
if candidate_rule_value > highest_candidate_rule_value[0]:
highest_candidate_rule_value = (candidate_rule_value, candidate_rule, p, n+p)
return highest_candidate_rule_value
def compute_binary_dl(rules, value, data, possible_conditions):
"""Binary DL: cost to encode rules for ONE class (target=value) + exceptions vs everything else.
Unlike the multi-class compute_ruleset_dl, this does NOT iterate over other classes — each
one-vs-rest sub-problem has its own independent DL, like wittgenstein."""
total_dl = 0
n = 2 * len(possible_conditions) # each column has both a True and False condition
# 1. Theory DL: cost of the rules themselves
for rule in rules:
if not rule:
continue
k = len(rule)
# Cohen 1995 (paper): log2(n+1) fixed-length encoding for k. Cost of "which k of n" is
# split into log2(n+1) for k itself + the combinatorial term for choosing which k conditions.
# This is stricter than wittgenstein's safe_log2(k) — it makes short rules more expensive
# and biases the build/delete stages toward keeping fewer rules.
theory_dl = (k * safe_log2(n/k) + (n-k) * safe_log2(n/(n-k)) + safe_log2(n+1)) * 0.5
total_dl += theory_dl
# 2. Coverage: rows where ANY rule for this class fires
covered = pd.Series(False, index=data.index)
for rule in rules:
if not rule:
continue
mask = pd.Series(True, index=data.index)
for cond in rule:
mask &= apply_cond_mask(data, cond)
covered |= mask
# 3. Exception DL: FP bits (covered rows that aren't target) + FN bits (target rows not covered)
fp_count = data[covered & (data['y'] != value)].shape[0]
total_covered = data[covered].shape[0]
fn_count = data[~covered & (data['y'] == value)].shape[0]
total_not_covered = data[~covered].shape[0]
fp_bits = safe_log2(safe_comb(total_covered, fp_count))
fn_bits = safe_log2(safe_comb(total_not_covered, fn_count))
total_dl += (fp_bits + fn_bits)
return total_dl
def stop_conditions(current_dl, best_dl):
if current_dl > best_dl + 64:
return True
return False
def grow_and_prune_rule(old_P, old_T, P, T, grow_set_new, prune_set, alternative_bool = False, old_rule = None, value = None, lambda_val=1, invalid_row_mask=None):
index = 0
rule = list(old_rule) if old_rule else []
P = grow_set_new[grow_set_new['y'] == value].shape[0]
T = grow_set_new.shape[0]
old_P = 0
old_T = 0
while not_perfect(old_P, old_T, P, T, index):
old_P = P
old_T = T
P, T, grow_set_new = grow(value, grow_set_new, rule, P, T, lambda_val, invalid_row_mask=invalid_row_mask)
index += 1
candidates = [rule[:i] for i in range(len(rule), -1, -1)]
prune_value, best_pruned_rule, final_P, final_T = prune(candidates, prune_set, value, alternative_bool, lambda_val)
return best_pruned_rule, final_P, final_T
def print_and_save_to_file(rule_set):
with open('final_ruleset.txt', 'w') as f:
for key, value in rule_set.items():
f.write(f"label: {key} rule: {value}\n")
def get_covered_mask(df, rule):
mask = pd.Series(True, index=df.index)
for cond in rule:
mask &= apply_cond_mask(df, cond)
return mask
def run_RIPPER_with_guidance_in_grow_and_prune(train_df=None, lambda_val=1):
"""One-vs-rest binary RIPPER, like wittgenstein. For each class, train an independent binary
classifier (target vs all others). Each binary sub-problem gets its own grow/prune split with
ALL non-target rows as negatives, so the prune set is stable even for rare classes.
If train_df (bool feature cols + a 'y' column) is given, train on it; otherwise fall back to
the module globals set by load_data()/the caller. train_df is NOT shuffled here — callers seed
numpy's global RNG and the per-class grow/prune splits handle randomisation.
Returns (rule_set, sorted_values)."""
global all_data, sorted_values, rule_set, possible_conditions
if train_df is not None:
all_data = train_df
value_counts = all_data["y"].value_counts().reset_index()
value_counts.columns = ["y", "count"]
sorted_values = value_counts.sort_values("count", ascending=True)["y"].tolist()
rule_set = {value: [] for value in sorted_values}
possible_conditions = all_data.columns.drop("y")
# MIN_PRECISION: at prediction time we still do multi-class voting, so the relevant baseline is
# the default class frequency in the full multi-class problem (not the 99% "majority not-target"
# of the binary sub-problem, which would block everything).
MIN_PRECISION = (all_data['y'] == sorted_values[-1]).sum() / len(all_data)
for value in sorted_values[:-1]:
# Per-class ontology mask, built once on all_data and reused by every grow call
# for this class (Stages 1, 2 replacement/revision, 3). Depends only on (value, all_data).
invalid_row_mask = build_invalid_row_mask(value, all_data)
# Per-class binary splits: target class vs all others. Three independent shuffles for Stage 2.
grow_set_local, prune_set_local = split_grow_prune_binary(all_data, value)
grow_set_replace_local, prune_set_replace_local = split_grow_prune_binary(all_data, value)
grow_set_revise_local, prune_set_revise_local = split_grow_prune_binary(all_data, value)
build_index = 1
# Initial DL: ruleset with no rules for this class (target rows are all FN, no FPs)
min_total_dl = compute_binary_dl([], value, all_data, possible_conditions)
curr_grow = grow_set_local.copy()
curr_prune = prune_set_local.copy()
# --- STAGE 1: BUILD ---
while True:
P = curr_grow[curr_grow['y'] == value].shape[0]
T = curr_grow.shape[0]
if P == 0:
break
rule, _, _ = grow_and_prune_rule(0, 0, P, T, curr_grow, curr_prune, False, None, value, lambda_val, invalid_row_mask)
if not rule:
break
# Reject rules whose precision on the FULL dataset is below the default class frequency.
full_mask = get_covered_mask(all_data, rule)
full_T = int(full_mask.sum())
full_P = int((all_data[full_mask]['y'] == value).sum())
if full_T > 0 and full_P / full_T < MIN_PRECISION:
break
new_dl = compute_binary_dl(rule_set[value] + [rule], value, all_data, possible_conditions)
if new_dl < min_total_dl:
min_total_dl = new_dl
if stop_conditions(new_dl, min_total_dl):
break
rule_set[value].append(rule)
grow_mask = get_covered_mask(curr_grow, rule)
prune_mask = get_covered_mask(curr_prune, rule)
curr_grow = curr_grow[~grow_mask]
curr_prune = curr_prune[~prune_mask]
build_index += 1
# --- STAGE 2: OPTIMIZE EACH RULE FOR THAT SPECIFIC RULE SET ---
# WEKA's JRip defaults to m_Optimizations = 2 — it runs the replacement/revision pass
# twice over each class's ruleset, generating fresh stratified splits each pass.
M_OPTIMIZATIONS = 2
for opt_pass in range(M_OPTIMIZATIONS):
# Fresh splits for this optimization pass.
grow_set_replace_local, prune_set_replace_local = split_grow_prune_binary(all_data, value)
grow_set_revise_local, prune_set_revise_local = split_grow_prune_binary(all_data, value)
for index in range(len(rule_set[value])):
current_rule = rule_set[value][index]
dl_orig = compute_binary_dl(rule_set[value], value, all_data, possible_conditions)
# Variant 1: Replacement
grow_set_new_replace = grow_set_replace_local.copy(deep=True)
prune_set_new_replace = prune_set_replace_local.copy(deep=True)
for new_index in range(len(rule_set[value])):
prev_clause = rule_set[value][new_index]
if new_index == index:
break
mask_grow = get_covered_mask(grow_set_new_replace, prev_clause)
mask_prune = get_covered_mask(prune_set_new_replace, prev_clause)
grow_set_new_replace = grow_set_new_replace[~mask_grow]
prune_set_new_replace = prune_set_new_replace[~mask_prune]
P = grow_set_new_replace[grow_set_new_replace['y'] == value].shape[0]
T = grow_set_new_replace.shape[0]
old_P = 0
old_T = 0
# Stage 2 uses accuracy (TP+TN)/(P+N) per the paper.
rule_replace, replace_P, _ = grow_and_prune_rule(old_P, old_T, P, T, grow_set_new_replace, prune_set_new_replace, True, None, value, lambda_val, invalid_row_mask)
dl_alt1 = compute_binary_dl(rule_set[value][:index] + [rule_replace] + rule_set[value][index + 1:], value, all_data, possible_conditions)
# Variant 2: Revision (start from current rule, add more conditions)
grow_set_new_revise = grow_set_revise_local.copy(deep=True)
prune_set_new_revise = prune_set_revise_local.copy(deep=True)
for new_index in range(len(rule_set[value])):
if new_index == index:
break
prev_clause = rule_set[value][new_index]
mask_grow = get_covered_mask(grow_set_new_revise, prev_clause)
mask_prune = get_covered_mask(prune_set_new_revise, prev_clause)
grow_set_new_revise = grow_set_new_revise[~mask_grow]
prune_set_new_revise = prune_set_new_revise[~mask_prune]
for antecedent in current_rule:
grow_set_new_revise = grow_set_new_revise[apply_cond_mask(grow_set_new_revise, antecedent)]
prune_set_new_revise = prune_set_new_revise[apply_cond_mask(prune_set_new_revise, antecedent)]
P = grow_set_new_revise[grow_set_new_revise['y'] == value].shape[0]
T = grow_set_new_revise.shape[0]
old_P = 0
old_T = 0
rule_revise, revise_P, _ = grow_and_prune_rule(old_P, old_T, P, T, grow_set_new_revise, prune_set_new_revise, True, list(current_rule), value, lambda_val, invalid_row_mask)
dl_alt2 = compute_binary_dl(rule_set[value][:index] + [rule_revise] + rule_set[value][index + 1:], value, all_data, possible_conditions)
if not rule_replace or replace_P == 0:
dl_alt1 = float('inf')
if not rule_revise or revise_P == 0:
dl_alt2 = float('inf')
if dl_orig <= dl_alt1 and dl_orig <= dl_alt2:
pass
elif dl_alt1 < dl_orig and dl_alt1 <= dl_alt2:
rule_set[value][index] = rule_replace
elif dl_alt2 < dl_orig and dl_alt2 < dl_alt1:
rule_set[value][index] = rule_revise
# --- STAGE 3: COVER REMAINING POSITIVES ---
covered_indices = set()
for rule in rule_set[value]:
mask = get_covered_mask(all_data, rule)
covered_indices.update(all_data[mask].index)
residual_positives = all_data[(all_data['y'] == value) & ~all_data.index.isin(covered_indices)]
if not residual_positives.empty:
# Fresh binary split: residual positives vs all original negatives.
negatives = all_data[all_data['y'] != value]
residual_data = shuffle_data(pd.concat([residual_positives, negatives]))
curr_grow, curr_prune = split_grow_prune_binary(residual_data, value)
build_index = 0
while True:
P = curr_grow[curr_grow['y'] == value].shape[0]
T = curr_grow.shape[0]
if P == 0:
break
rule, _, _ = grow_and_prune_rule(0, 0, P, T, curr_grow, curr_prune, False, None, value, lambda_val, invalid_row_mask)
if not rule:
break
full_mask = get_covered_mask(all_data, rule)
full_T = int(full_mask.sum())
full_P = int((all_data[full_mask]['y'] == value).sum())
if full_T > 0 and full_P / full_T < MIN_PRECISION:
break
new_dl = compute_binary_dl(rule_set[value] + [rule], value, all_data, possible_conditions)
if new_dl < min_total_dl:
min_total_dl = new_dl
if stop_conditions(new_dl, min_total_dl):
break
rule_set[value].append(rule)
grow_mask = get_covered_mask(curr_grow, rule)
prune_mask = get_covered_mask(curr_prune, rule)
curr_grow = curr_grow[~grow_mask]
curr_prune = curr_prune[~prune_mask]
build_index += 1
# --- STAGE 4: DELETE RULES THAT INCREASE DL (per binary sub-problem) ---
# Each class has its own independent binary DL — delete rules that don't pay for themselves
# against the target-vs-rest objective only.
current_dl = compute_binary_dl(rule_set[value], value, all_data, possible_conditions)
for rule in list(rule_set[value]):
idx = rule_set[value].index(rule)
candidate_rules = rule_set[value][:idx] + rule_set[value][idx+1:]
candidate_dl = compute_binary_dl(candidate_rules, value, all_data, possible_conditions)
# WEKA uses strict `<` here: a rule is only deleted if removing it STRICTLY lowers DL.
# With `<=`, ties would discard rules that pay for themselves exactly — costing recall
# without any DL benefit.
if candidate_dl < current_dl:
current_dl = candidate_dl
rule_set[value] = rule_set[value][:idx] + rule_set[value][idx+1:]
return rule_set, sorted_values

View file

@ -0,0 +1,447 @@
# Binary one-vs-rest version of jrip_my_version_with_ai.py.
# For each class (rarest first), this trains a binary RIPPER: target class vs everything else.
# At prediction time, the first class whose rule fires wins; if no rule fires, fall back to the
# most-common class — same scheme as evaluate_ruleset.py.
#
# COMBINED variant: applies the ontology penalty at ALL THREE places at once —
# (1) grow() : subtracts lambda*viol from each candidate literal's FOIL gain,
# (2) prune() : subtracts lambda*viol from the pruning metric,
# (3) compute_binary_dl() : adds lambda*viol to the description length used by the
# accept/reject/optimize/delete (MDL) stages.
# i.e. "onto grow and prune" + the "onto MDL" DL penalty stacked together.
from math import log, comb, log2
from pathlib import Path
import pandas as pd
from pandas import DataFrame
from ontology_guidance.ontology_guidance import compute_rule_viol, compute_ontology_viol
from helper_functions.build_invalid_row_mask import build_invalid_row_mask
EXCEL_DIR = Path(__file__).parent / "excels"
def shuffle_data(df):
df_new = df.sample(frac=1)
return df_new
def load_data(dataset_suffix="less_classes"):
"""Load excels/X_{suffix}.xlsx + y_{suffix}.xlsx and set the module globals
(all_data, sorted_values, rule_set, possible_conditions) that the run
function reads. Returns nothing — mutates module state."""
global all_data, sorted_values, rule_set, possible_conditions
df = pd.read_excel(EXCEL_DIR / f'X_{dataset_suffix}.xlsx', index_col=0)
y_df = pd.read_excel(EXCEL_DIR / f'y_{dataset_suffix}.xlsx')
df['y'] = y_df['y'].values
all_data = shuffle_data(df)
value_counts = all_data["y"].value_counts().reset_index()
value_counts.columns = ["y", "count"]
sorted_values = value_counts.sort_values("count", ascending=True)["y"].tolist()
rule_set = {value: [] for value in sorted_values}
possible_conditions = all_data.columns.drop('y')
return all_data
def safe_log2(x):
return log2(x) if x > 0 else 0
def safe_log(x):
return log(x) if x > 0 else 0
def safe_comb(n, k):
if n < 0 or k < 0 or k > n:
return 1
return comb(n, k)
def apply_cond_mask(df, cond):
"""Return a boolean Series: True where rows satisfy cond.
Negative conditions are represented with a leading '!' (e.g. '!col' means col==False)."""
if cond.startswith('!'):
return df[cond[1:]] == False
return df[cond] == True
def split_grow_prune_binary(data, target_value):
"""Binary 2/3 - 1/3 split, stratified on target vs not-target (matches wittgenstein's one-vs-rest).
Positives and negatives are each shuffled then split, so both grow and prune sets contain
representative numbers from both classes."""
positives = data[data['y'] == target_value].sample(frac=1)
negatives = data[data['y'] != target_value].sample(frac=1)
pos_split = int(len(positives) * (2/3))
neg_split = int(len(negatives) * (2/3))
if len(positives) <= 2:
grow = pd.concat([positives, negatives.iloc[:neg_split]])
prune = pd.concat([positives, negatives.iloc[neg_split:]])
else:
grow = pd.concat([positives.iloc[:pos_split], negatives.iloc[:neg_split]])
prune = pd.concat([positives.iloc[pos_split:], negatives.iloc[neg_split:]])
return shuffle_data(grow), shuffle_data(prune)
def not_perfect(old_P, old_T, P, T, index):
best_gain = 0
if P != 0 and T != 0 and old_P != 0 and old_T != 0:
best_gain = P*(safe_log(P/T)-safe_log(old_P/old_T))
if P != 0 and T != 0 and P == T:
return False # perfect, stop
if best_gain < 0 or (best_gain == 0 and index > 0):
return False
return True
def grow(correct_y, all_x, rule, P, T, lambda_val, invalid_row_mask):
max_info_gain_tuples = (0, '', pd.DataFrame())
final_p = 0
final_t = 0
for col in all_x.columns.drop('y'):
for val in [True, False]:
cond_name = ('!' + col) if not val else col
if cond_name in rule: # skip conditions already in the rule
continue
candidate_rule_outcome = all_x[apply_cond_mask(all_x, cond_name)]
t = candidate_rule_outcome.shape[0]
new_data = candidate_rule_outcome[candidate_rule_outcome['y'] == correct_y]
p = new_data.shape[0]
if p == 0 or t == 0:
information_gain = 0
else:
information_gain = p * (safe_log(p/t) - safe_log(P/T))
viol = compute_rule_viol(list(rule + [cond_name]), correct_y, all_x, invalid_row_mask)
information_gain = information_gain - lambda_val * viol
# WEKA tie-breaking: when two conditions have the same info_gain, prefer the
# MORE SPECIFIC one (smaller `t` = covers fewer rows). With strict `>` the first
# column iterated wins ties — that's an artifact of column order in the dataframe,
# not a meaningful choice. Tie-breaking on smaller `t` produces tighter rules and
# makes grow's output deterministic w.r.t. column order.
better = information_gain > max_info_gain_tuples[0]
tie_break = (information_gain == max_info_gain_tuples[0]
and information_gain > 0
and t < final_t)
if better or tie_break:
final_new_data = new_data.copy(deep=True)
final_p = p
final_t = t
max_info_gain_tuples = (information_gain, cond_name, final_new_data)
condition = max_info_gain_tuples[1]
final_new_data = max_info_gain_tuples[2]
if not condition:
return P, T, all_x
rule.append(condition)
final_new_data = all_x[apply_cond_mask(all_x, condition)]
return final_p, final_t, final_new_data
def prune(candidates, prune_set, value, alternative_bool, lambda_val):
""" Incrementally prune each rule and allow the pruning of any final sequences of the antecedents and the pruning metric is: (p+1)/(p+n+2)"""
highest_candidate_rule_value = (0, [], 0, 1)
invalid_row_mask_prune = build_invalid_row_mask(value, prune_set)
all_positives = prune_set[prune_set['y'] == value].shape[0]
all_negatives = prune_set[prune_set['y'] != value].shape[0]
if all_positives + all_negatives == 0:
return highest_candidate_rule_value
for candidate_rule in candidates:
prune_set_filtered = prune_set
n_onto = compute_rule_viol(candidate_rule, value, prune_set, invalid_row_mask_prune)
if alternative_bool:
for antecedent in candidate_rule:
prune_set_filtered = prune_set_filtered.loc[apply_cond_mask(prune_set_filtered, antecedent)]
not_covered = prune_set.loc[~prune_set.index.isin(prune_set_filtered.index)]
true_negatives = not_covered.loc[not_covered['y'] != value].shape[0]
true_positives = prune_set_filtered.loc[prune_set_filtered['y'] == value].shape[0]
all_positives = prune_set.loc[prune_set['y'] == value].shape[0]
all_negatives = prune_set.loc[prune_set['y'] != value].shape[0]
candidate_rule_value = ((true_positives + true_negatives) - (lambda_val * n_onto)) / (all_positives + all_negatives)
if candidate_rule_value > highest_candidate_rule_value[0]:
highest_candidate_rule_value = (candidate_rule_value, candidate_rule, true_positives, true_negatives+true_positives)
else:
for antecedent in candidate_rule:
prune_set_filtered = prune_set_filtered.loc[apply_cond_mask(prune_set_filtered, antecedent)]
n = prune_set_filtered.loc[prune_set_filtered['y'] != value].shape[0]
p = prune_set_filtered.loc[prune_set_filtered['y'] == value].shape[0]
#candidate_rule_value = (p+1)/(p+n+2)
candidate_rule_value = (p + 1) / (p + n + lambda_val * n_onto + 2)
if candidate_rule_value > highest_candidate_rule_value[0]:
highest_candidate_rule_value = (candidate_rule_value, candidate_rule, p, n+p)
return highest_candidate_rule_value
def compute_binary_dl(rules, value, data, possible_conditions, lambda_val=0):
"""Binary DL: cost to encode rules for ONE class (target=value) + exceptions vs everything else.
Unlike the multi-class compute_ruleset_dl, this does NOT iterate over other classes — each
one-vs-rest sub-problem has its own independent DL, like wittgenstein.
COMBINED variant: a lambda*viol ontology penalty is ALSO added to the DL (the "onto MDL"
term), on top of the grow/prune penalties — so all four MDL stages (accept, replace/revise,
residual add, delete) see the ontology cost too."""
total_dl = 0
n = 2 * len(possible_conditions) # each column has both a True and False condition
# 1. Theory DL: cost of the rules themselves
for rule in rules:
if not rule:
continue
k = len(rule)
# Cohen 1995 (paper): log2(n+1) fixed-length encoding for k. Cost of "which k of n" is
# split into log2(n+1) for k itself + the combinatorial term for choosing which k conditions.
# This is stricter than wittgenstein's safe_log2(k) — it makes short rules more expensive
# and biases the build/delete stages toward keeping fewer rules.
theory_dl = (k * safe_log2(n/k) + (n-k) * safe_log2(n/(n-k)) + safe_log2(n+1)) * 0.5
total_dl += theory_dl
# 2. Coverage: rows where ANY rule for this class fires
covered = pd.Series(False, index=data.index)
for rule in rules:
if not rule:
continue
mask = pd.Series(True, index=data.index)
for cond in rule:
mask &= apply_cond_mask(data, cond)
covered |= mask
# 3. Exception DL: FP bits (covered rows that aren't target) + FN bits (target rows not covered)
fp_count = data[covered & (data['y'] != value)].shape[0]
total_covered = data[covered].shape[0]
fn_count = data[~covered & (data['y'] == value)].shape[0]
total_not_covered = data[~covered].shape[0]
fp_bits = safe_log2(safe_comb(total_covered, fp_count))
fn_bits = safe_log2(safe_comb(total_not_covered, fn_count))
total_dl += (fp_bits + fn_bits)
# COMBINED: ontology violation penalty on the description length (the "onto MDL" term).
ont_violations = compute_ontology_viol(rules, value, data)
total_dl += ont_violations * lambda_val
return total_dl
def stop_conditions(current_dl, best_dl):
if current_dl > best_dl + 64:
return True
return False
def grow_and_prune_rule(old_P, old_T, P, T, grow_set_new, prune_set, alternative_bool = False, old_rule = None, value = None, lambda_val=1, invalid_row_mask=None):
index = 0
rule = list(old_rule) if old_rule else []
P = grow_set_new[grow_set_new['y'] == value].shape[0]
T = grow_set_new.shape[0]
old_P = 0
old_T = 0
while not_perfect(old_P, old_T, P, T, index):
old_P = P
old_T = T
P, T, grow_set_new = grow(value, grow_set_new, rule, P, T, lambda_val, invalid_row_mask=invalid_row_mask)
index += 1
candidates = [rule[:i] for i in range(len(rule), -1, -1)]
prune_value, best_pruned_rule, final_P, final_T = prune(candidates, prune_set, value, alternative_bool, lambda_val)
return best_pruned_rule, final_P, final_T
def print_and_save_to_file(rule_set):
with open('final_ruleset.txt', 'w') as f:
for key, value in rule_set.items():
f.write(f"label: {key} rule: {value}\n")
def get_covered_mask(df, rule):
mask = pd.Series(True, index=df.index)
for cond in rule:
mask &= apply_cond_mask(df, cond)
return mask
def run_RIPPER_with_guidance_in_grow_prune_mdl(train_df=None, lambda_val=1):
"""One-vs-rest binary RIPPER, like wittgenstein. For each class, train an independent binary
classifier (target vs all others). Each binary sub-problem gets its own grow/prune split with
ALL non-target rows as negatives, so the prune set is stable even for rare classes.
If train_df (bool feature cols + a 'y' column) is given, train on it; otherwise fall back to
the module globals set by load_data()/the caller. train_df is NOT shuffled here — callers seed
numpy's global RNG and the per-class grow/prune splits handle randomisation.
Returns (rule_set, sorted_values)."""
global all_data, sorted_values, rule_set, possible_conditions
if train_df is not None:
all_data = train_df
value_counts = all_data["y"].value_counts().reset_index()
value_counts.columns = ["y", "count"]
sorted_values = value_counts.sort_values("count", ascending=True)["y"].tolist()
rule_set = {value: [] for value in sorted_values}
possible_conditions = all_data.columns.drop("y")
# MIN_PRECISION: at prediction time we still do multi-class voting, so the relevant baseline is
# the default class frequency in the full multi-class problem (not the 99% "majority not-target"
# of the binary sub-problem, which would block everything).
MIN_PRECISION = (all_data['y'] == sorted_values[-1]).sum() / len(all_data)
for value in sorted_values[:-1]:
# Per-class ontology mask, built once on all_data and reused by every grow call
# for this class (Stages 1, 2 replacement/revision, 3). Depends only on (value, all_data).
invalid_row_mask = build_invalid_row_mask(value, all_data)
# Per-class binary splits: target class vs all others. Three independent shuffles for Stage 2.
grow_set_local, prune_set_local = split_grow_prune_binary(all_data, value)
grow_set_replace_local, prune_set_replace_local = split_grow_prune_binary(all_data, value)
grow_set_revise_local, prune_set_revise_local = split_grow_prune_binary(all_data, value)
build_index = 1
# Initial DL: ruleset with no rules for this class (target rows are all FN, no FPs)
min_total_dl = compute_binary_dl([], value, all_data, possible_conditions, lambda_val)
curr_grow = grow_set_local.copy()
curr_prune = prune_set_local.copy()
# --- STAGE 1: BUILD ---
while True:
P = curr_grow[curr_grow['y'] == value].shape[0]
T = curr_grow.shape[0]
if P == 0:
break
rule, _, _ = grow_and_prune_rule(0, 0, P, T, curr_grow, curr_prune, False, None, value, lambda_val, invalid_row_mask)
if not rule:
break
# Reject rules whose precision on the FULL dataset is below the default class frequency.
full_mask = get_covered_mask(all_data, rule)
full_T = int(full_mask.sum())
full_P = int((all_data[full_mask]['y'] == value).sum())
if full_T > 0 and full_P / full_T < MIN_PRECISION:
break
new_dl = compute_binary_dl(rule_set[value] + [rule], value, all_data, possible_conditions, lambda_val)
if new_dl < min_total_dl:
min_total_dl = new_dl
if stop_conditions(new_dl, min_total_dl):
break
rule_set[value].append(rule)
grow_mask = get_covered_mask(curr_grow, rule)
prune_mask = get_covered_mask(curr_prune, rule)
curr_grow = curr_grow[~grow_mask]
curr_prune = curr_prune[~prune_mask]
build_index += 1
# --- STAGE 2: OPTIMIZE EACH RULE FOR THAT SPECIFIC RULE SET ---
# WEKA's JRip defaults to m_Optimizations = 2 — it runs the replacement/revision pass
# twice over each class's ruleset, generating fresh stratified splits each pass.
M_OPTIMIZATIONS = 2
for opt_pass in range(M_OPTIMIZATIONS):
# Fresh splits for this optimization pass.
grow_set_replace_local, prune_set_replace_local = split_grow_prune_binary(all_data, value)
grow_set_revise_local, prune_set_revise_local = split_grow_prune_binary(all_data, value)
for index in range(len(rule_set[value])):
current_rule = rule_set[value][index]
dl_orig = compute_binary_dl(rule_set[value], value, all_data, possible_conditions, lambda_val)
# Variant 1: Replacement
grow_set_new_replace = grow_set_replace_local.copy(deep=True)
prune_set_new_replace = prune_set_replace_local.copy(deep=True)
for new_index in range(len(rule_set[value])):
prev_clause = rule_set[value][new_index]
if new_index == index:
break
mask_grow = get_covered_mask(grow_set_new_replace, prev_clause)
mask_prune = get_covered_mask(prune_set_new_replace, prev_clause)
grow_set_new_replace = grow_set_new_replace[~mask_grow]
prune_set_new_replace = prune_set_new_replace[~mask_prune]
P = grow_set_new_replace[grow_set_new_replace['y'] == value].shape[0]
T = grow_set_new_replace.shape[0]
old_P = 0
old_T = 0
# Stage 2 uses accuracy (TP+TN)/(P+N) per the paper.
rule_replace, replace_P, _ = grow_and_prune_rule(old_P, old_T, P, T, grow_set_new_replace, prune_set_new_replace, True, None, value, lambda_val, invalid_row_mask)
dl_alt1 = compute_binary_dl(rule_set[value][:index] + [rule_replace] + rule_set[value][index + 1:], value, all_data, possible_conditions, lambda_val)
# Variant 2: Revision (start from current rule, add more conditions)
grow_set_new_revise = grow_set_revise_local.copy(deep=True)
prune_set_new_revise = prune_set_revise_local.copy(deep=True)
for new_index in range(len(rule_set[value])):
if new_index == index:
break
prev_clause = rule_set[value][new_index]
mask_grow = get_covered_mask(grow_set_new_revise, prev_clause)
mask_prune = get_covered_mask(prune_set_new_revise, prev_clause)
grow_set_new_revise = grow_set_new_revise[~mask_grow]
prune_set_new_revise = prune_set_new_revise[~mask_prune]
for antecedent in current_rule:
grow_set_new_revise = grow_set_new_revise[apply_cond_mask(grow_set_new_revise, antecedent)]
prune_set_new_revise = prune_set_new_revise[apply_cond_mask(prune_set_new_revise, antecedent)]
P = grow_set_new_revise[grow_set_new_revise['y'] == value].shape[0]
T = grow_set_new_revise.shape[0]
old_P = 0
old_T = 0
rule_revise, revise_P, _ = grow_and_prune_rule(old_P, old_T, P, T, grow_set_new_revise, prune_set_new_revise, True, list(current_rule), value, lambda_val, invalid_row_mask)
dl_alt2 = compute_binary_dl(rule_set[value][:index] + [rule_revise] + rule_set[value][index + 1:], value, all_data, possible_conditions, lambda_val)
if not rule_replace or replace_P == 0:
dl_alt1 = float('inf')
if not rule_revise or revise_P == 0:
dl_alt2 = float('inf')
if dl_orig <= dl_alt1 and dl_orig <= dl_alt2:
pass
elif dl_alt1 < dl_orig and dl_alt1 <= dl_alt2:
rule_set[value][index] = rule_replace
elif dl_alt2 < dl_orig and dl_alt2 < dl_alt1:
rule_set[value][index] = rule_revise
# --- STAGE 3: COVER REMAINING POSITIVES ---
covered_indices = set()
for rule in rule_set[value]:
mask = get_covered_mask(all_data, rule)
covered_indices.update(all_data[mask].index)
residual_positives = all_data[(all_data['y'] == value) & ~all_data.index.isin(covered_indices)]
if not residual_positives.empty:
# Fresh binary split: residual positives vs all original negatives.
negatives = all_data[all_data['y'] != value]
residual_data = shuffle_data(pd.concat([residual_positives, negatives]))
curr_grow, curr_prune = split_grow_prune_binary(residual_data, value)
build_index = 0
while True:
P = curr_grow[curr_grow['y'] == value].shape[0]
T = curr_grow.shape[0]
if P == 0:
break
rule, _, _ = grow_and_prune_rule(0, 0, P, T, curr_grow, curr_prune, False, None, value, lambda_val, invalid_row_mask)
if not rule:
break
full_mask = get_covered_mask(all_data, rule)
full_T = int(full_mask.sum())
full_P = int((all_data[full_mask]['y'] == value).sum())
if full_T > 0 and full_P / full_T < MIN_PRECISION:
break
new_dl = compute_binary_dl(rule_set[value] + [rule], value, all_data, possible_conditions, lambda_val)
if new_dl < min_total_dl:
min_total_dl = new_dl
if stop_conditions(new_dl, min_total_dl):
break
rule_set[value].append(rule)
grow_mask = get_covered_mask(curr_grow, rule)
prune_mask = get_covered_mask(curr_prune, rule)
curr_grow = curr_grow[~grow_mask]
curr_prune = curr_prune[~prune_mask]
build_index += 1
# --- STAGE 4: DELETE RULES THAT INCREASE DL (per binary sub-problem) ---
# Each class has its own independent binary DL — delete rules that don't pay for themselves
# against the target-vs-rest objective only.
current_dl = compute_binary_dl(rule_set[value], value, all_data, possible_conditions, lambda_val)
for rule in list(rule_set[value]):
idx = rule_set[value].index(rule)
candidate_rules = rule_set[value][:idx] + rule_set[value][idx+1:]
candidate_dl = compute_binary_dl(candidate_rules, value, all_data, possible_conditions, lambda_val)
# WEKA uses strict `<` here: a rule is only deleted if removing it STRICTLY lowers DL.
# With `<=`, ties would discard rules that pay for themselves exactly — costing recall
# without any DL benefit.
if candidate_dl < current_dl:
current_dl = candidate_dl
rule_set[value] = rule_set[value][:idx] + rule_set[value][idx+1:]
return rule_set, sorted_values

606
run_all_experiments.py Normal file
View file

@ -0,0 +1,606 @@
"""
Run ALL ontology-guidance experiments in one go and save everything.
Sweep: 4 methods x lambda in {3, 4} x pollution levels [0..100] x N_REPEATS repeats
of stratified 5-fold CV. Each repeat is a different row->fold partition (and a different
pollution/RIPPER seed, both keyed off `run`), so repeats are genuine resamples. For each
(repeat, fold) a RIPPER variant is trained and, under runtime ontology guidance OFF and ON,
per-fold metrics PLUS the raw out-of-fold predictions are recorded:
train_accuracy, test_accuracy, test_f1_macro, test_f1_weighted, invalid_rate
where invalid_rate = fraction of TEST predictions that are ontologically invalid
for the row's asset.
To pull a small guidance effect out of fold/rare-class noise we then (a) POOL each repeat's
five out-of-fold prediction sets into ONE macro-F1 over all rows (every row is predicted
exactly once per repeat, so rare classes contribute all their instances at once — far
steadier than averaging five tiny per-fold F1s), and (b) run a PAIRED guided-vs-unguided
test per pollution level (all methods share each fold's data/pollution/seed, so the fold
noise cancels in the difference).
Outputs (under ./results):
metrics_long.csv master per-fold metrics, resumable; one row per (lambda, level, run, method, runtime).
oof_predictions.csv raw out-of-fold predictions; one row per held-out test row x method x runtime.
pooled_metrics.csv pooled OOF score, one row per (lambda, level, repeat, method, runtime).
paired_tests.csv guided-vs-unguided paired macro-F1 tests per (lambda, level, runtime).
summary_runtimeOFF.csv MAIN summary: mean/std over the runs, runtime guidance OFF.
runtime_ON_results.csv EXTRA: same summary but for runtime guidance ON.
figures/lambda{3,4}/<metric>.png per-fold plots (runtime OFF, one line per method).
figures/pooled/lambda{3,4}/<metric>.png POOLED-OOF plots (runtime OFF) — the steadier curves.
figures/runtime_ON/<metric>_lambda{}.png EXTRA plots (runtime ON).
Plus a separate "comparing to prof" experiment on the CLEAN data (no pollution,
runtime OFF): the 4 custom variants vs Wittgenstein and Weka JRip, over the same
10 runs, scored on test accuracy + macro & weighted F1:
prof_comparison_clean.csv per (run, method) rows.
prof_comparison_summary.csv mean/std per method.
figures/prof_comparison/{test_accuracy,test_f1_macro,test_f1_weighted}.png
The main graphs are runtime-OFF only (clean); the runtime-ON results live in
their own clearly-named file/figures so they don't clutter the main plots.
Resumable: re-running skips any (lambda, level, run, method) already in
metrics_long.csv, so if it crashes (or you Ctrl-C) just run it again to continue.
Reproducible: pollution is deterministic and np.random.seed(run) is set before
each fit (same data/shuffle across lambdas), so reruns give identical numbers.
Usage:
python run_all_experiments.py # compute (resuming) main sweep + prof comparison + plot
python run_all_experiments.py --fresh # back up old CSV(s) and recompute from scratch
python run_all_experiments.py --plot-only # only (re)build summaries + figures from the CSVs
python run_all_experiments.py --prof-only # only the clean-data 'comparing to prof' experiment + plots
"""
import argparse
import textwrap
from pathlib import Path
import matplotlib
matplotlib.use("Agg") # headless: save figures, never open a window
import matplotlib.pyplot as plt
import matplotlib.ticker as mticker
import numpy as np
import pandas as pd
import seaborn as sns
from sklearn.metrics import f1_score
from sklearn.model_selection import RepeatedStratifiedKFold
from scipy.stats import wilcoxon, ttest_rel
# ── reused project pieces (lightweight imports only) ──────────────────────────
from RIPPER_no_onto_guid import run_RIPPER_without_guidance
from RIPPER_onto_guid import run_RIPPER_with_guidance
from RIPPER_onto_guid_grow import run_RIPPER_with_guidance_in_grow
from RIPPER_onto_guid_grow_and_prune import run_RIPPER_with_guidance_in_grow_and_prune
from evaluate_ruleset import evaluate_ruleset, ABSTAIN
from oud_maak_verv_dataset import pollute_train_labels_my_version
from ontology_guidance.get_onto_guidance_actions_per_asset import get_ontology_info_actions_per_asset
# prof baselines (Wittgenstein + Weka JRip) — reuse the proven helpers from the comparison module
import contextlib, io, os, re, subprocess, tempfile
from sklearn.multiclass import OneVsRestClassifier
from final_compare_accuracies_no_vervuiling import _SklearnRIPPER, _write_arff, WEKA_JAR
# ── config (overridable via EXP_* env vars, handy for a quick smoke test) ─────
def _env_int_list(name, default):
raw = os.environ.get(name)
return [int(x) for x in raw.split(",")] if raw else default
LAMBDAS = _env_int_list("EXP_LAMBDAS", [3, 4]) # ontology penalty weights (alpha)
POLLUTION_LEVELS = _env_int_list("EXP_LEVELS", [0, 10, 20, 30, 40, 50, 60, 70, 80, 90, 100])
N_FOLDS = 5
# Repeated stratified 5-fold CV. Each repeat is a DIFFERENT row->fold partition (and a
# different pollution/RIPPER seed, since both key off `run`), so repeats are real resamples.
# We summarise each repeat by ONE pooled out-of-fold macro-F1, then use the repeats for the
# spread/CI and the paired test. More repeats -> tighter bands + more paired power.
N_REPEATS = int(os.environ.get("EXP_N_REPEATS", "50"))
SPLIT_SEED = 42 # fixed splits, identical for every method/lambda
EXCEL_DIR = Path("excels")
OUTDIR = Path(os.environ.get("EXP_OUTDIR", "results"))
RESULTS_CSV = OUTDIR / "metrics_long.csv" # master per-fold metrics — resumable
OOF_CSV = OUTDIR / "oof_predictions.csv" # raw out-of-fold predictions (per held-out row)
POOLED_CSV = OUTDIR / "pooled_metrics.csv" # pooled OOF score per (lambda,level,repeat,method,runtime)
PAIRED_CSV = OUTDIR / "paired_tests.csv" # guided-vs-unguided paired tests per (lambda,level,runtime)
SUMMARY_CSV = OUTDIR / "summary_runtimeOFF.csv" # MAIN summary (runtime OFF)
RUNTIME_ON_CSV = OUTDIR / "runtime_ON_results.csv" # EXTRA summary (runtime ON)
FIGDIR = OUTDIR / "figures"
METRICS = ["train_accuracy", "test_accuracy", "test_f1_macro", "test_f1_weighted", "invalid_rate", "abstain_rate"]
COLUMNS = ["lambda", "percentage", "run", "method", "runtime", *METRICS]
OOF_COLUMNS = ["lambda", "percentage", "repeat", "fold", "method", "runtime",
"idx", "asset", "y_true", "y_pred"]
GUIDED_METHODS = ["onto MDL", "onto grow", "onto grow and prune"]
def repeat_fold(run):
"""Map the 1-based `run` index to (repeat, fold) for N_FOLDS-fold repeated CV."""
return (run - 1) // N_FOLDS + 1, (run - 1) % N_FOLDS + 1
# ── "comparing to prof" config (clean dataset, no pollution, runtime guidance OFF) ──
PROF_LAMBDA = 4 # guided variants' lambda for this clean-data run
PROF_CSV = OUTDIR / "prof_comparison_clean.csv"
PROF_SUMMARY_CSV = OUTDIR / "prof_comparison_summary.csv"
PROF_FIGDIR = FIGDIR / "prof_comparison"
PROF_METRICS = ["test_accuracy", "test_f1_macro", "test_f1_weighted"]
PROF_COLUMNS = ["run", "method", *PROF_METRICS]
PROF_ORDER = ["no guidance", "onto MDL", "onto grow", "onto grow and prune",
"Wittgenstein", "WEKA"]
def build_methods(lam):
"""The 4 RIPPER variants for a given lambda (no-guidance ignores lambda)."""
return [
("no guidance", run_RIPPER_without_guidance, {}),
("onto MDL", run_RIPPER_with_guidance, {"lambda_val": lam}),
("onto grow", run_RIPPER_with_guidance_in_grow, {"lambda_val": lam}),
("onto grow and prune", run_RIPPER_with_guidance_in_grow_and_prune, {"lambda_val": lam}),
]
# ── small helpers (copied from the notebook to keep this file self-contained) ─
def load_clean_dummies():
"""Clean less_classes dataset, one-hot encoded. Returns (X, y, all_classes)."""
df = pd.read_excel(EXCEL_DIR / "non_dummy_columns_df_less_classes.xlsx")
X = pd.get_dummies(df.drop("y", axis=1)).astype(bool)
y = df["y"]
return X, y, sorted(y.unique().tolist())
def make_fold_dfs(X, y, train_idx, test_idx):
"""(train_df, test_df) for one fold: bool feature cols + a 'y' column."""
train = X.iloc[train_idx].copy(); train["y"] = y.iloc[train_idx].values
test = X.iloc[test_idx].copy(); test["y"] = y.iloc[test_idx].values
return train, test
def _row_asset(row):
"""Asset of a one-hot row: the assetType_* column that is True (or None)."""
for col in row.index:
if col.startswith("assetType_") and row[col] == True:
return col.replace("assetType_", "")
return None
def ontology_invalid_rate(result_df):
"""Fraction of predictions whose action is ontologically INVALID for the row's asset.
Invalid only when the asset has a known, non-empty valid-action set and y_pred is not
in it (same notion the runtime guidance uses); unknown asset -> not counted."""
def _invalid(row):
if str(row["y_pred"]) == ABSTAIN: # abstention is not an invalid action (see abstain_rate)
return False
asset = _row_asset(row)
if asset is None:
return False
valid = {a.strip().lower() for a in get_ontology_info_actions_per_asset(asset)}
return bool(valid) and str(row["y_pred"]).strip().lower() not in valid
return result_df.apply(_invalid, axis=1).mean()
# ── checkpoint I/O ────────────────────────────────────────────────────────────
def load_done():
"""Set of (lambda, percentage, run, method) already present in the results CSV."""
if not RESULTS_CSV.exists():
return set()
df = pd.read_csv(RESULTS_CSV)
return {(int(r.lambda_), int(r.percentage), int(r.run), str(r.method))
for r in df.rename(columns={"lambda": "lambda_"}).itertuples()}
def append_rows(rows):
"""Append rows (list of dicts) to the CSV, writing the header only once."""
pd.DataFrame(rows, columns=COLUMNS).to_csv(
RESULTS_CSV, mode="a", header=not RESULTS_CSV.exists(), index=False
)
def append_oof(rows):
"""Append per-row out-of-fold predictions to OOF_CSV (header written once)."""
if not rows:
return
pd.DataFrame(rows, columns=OOF_COLUMNS).to_csv(
OOF_CSV, mode="a", header=not OOF_CSV.exists(), index=False
)
# ── experiment ────────────────────────────────────────────────────────────────
def run_experiments():
OUTDIR.mkdir(parents=True, exist_ok=True)
X, y, _ = load_clean_dummies()
rskf = RepeatedStratifiedKFold(n_splits=N_FOLDS, n_repeats=N_REPEATS, random_state=SPLIT_SEED)
done = load_done()
print(f"{len(done)} (lambda,level,run,method) units already done — skipping those.\n")
for level in POLLUTION_LEVELS:
for run, (train_idx, test_idx) in enumerate(rskf.split(X, y), start=1):
rep, fld = repeat_fold(run)
# split + pollution computed once per (level, run); reused for every lambda
train_df, test_df = make_fold_dfs(X, y, train_idx, test_idx)
train_df, _ = pollute_train_labels_my_version(train_df, level, run)
for lam in LAMBDAS:
for name, fn, kw in build_methods(lam):
if (lam, level, run, name) in done:
continue
np.random.seed(run) # reproducible + fair paired comparison across lambdas
rule_set, sorted_values = fn(train_df.copy(), **kw)
batch, oof_batch = [], []
for runtime in ("OFF", "ON"):
rt = runtime == "ON"
_, train_acc = evaluate_ruleset(rule_set, train_df.copy(), sorted_values, rt)
test_res, test_acc = evaluate_ruleset(rule_set, test_df.copy(), sorted_values, rt)
batch.append({
"lambda": lam, "percentage": level, "run": run,
"method": name, "runtime": runtime,
"train_accuracy": train_acc,
"test_accuracy": test_acc,
"test_f1_macro": f1_score(test_res["y"], test_res["y_pred"],
average="macro", zero_division=0),
"test_f1_weighted": f1_score(test_res["y"], test_res["y_pred"],
average="weighted", zero_division=0),
"invalid_rate": ontology_invalid_rate(test_res),
"abstain_rate": float((test_res["y_pred"] == ABSTAIN).mean()),
})
# raw out-of-fold predictions, for pooled-per-repeat scoring later
for idx, row in test_res.iterrows():
oof_batch.append({
"lambda": lam, "percentage": level, "repeat": rep, "fold": fld,
"method": name, "runtime": runtime, "idx": int(idx),
"asset": _row_asset(row), "y_true": row["y"], "y_pred": row["y_pred"],
})
append_rows(batch) # flush after each method -> crash-safe
append_oof(oof_batch)
off, on = batch
print(f"λ={lam} | level {level:>3} | run {run:>2} | {name:<20} "
f"test_acc OFF={off['test_accuracy']:.3f} ON={on['test_accuracy']:.3f} | "
f"invalid% OFF={off['invalid_rate']:.1%} ON={on['invalid_rate']:.1%}")
print("\nAll experiments done.")
# ── plots + summaries (also runnable standalone via --plot-only) ──────────────
PRETTY = {"train_accuracy": "Train accuracy", "test_accuracy": "Test accuracy",
"test_f1_macro": "Test macro-F1", "test_f1_weighted": "Test weighted-F1",
"invalid_rate": "% predictions ontologically invalid",
"abstain_rate": "% predictions abstained (no valid action)"}
def _lineplot(sub, metric, title, out):
"""One clean line per method (averaged over runs via CI band), saved to `out`."""
plt.figure(figsize=(9, 6))
ax = sns.lineplot(data=sub, x="percentage", y=metric, hue="method",
marker="o", markersize=7, linewidth=2.2, errorbar="ci")
# invalid_rate / abstain_rate have long y-label wording, so their composed titles run off
# the figure — wrap onto multiple lines and shrink the font just for those two metrics.
if metric in ("invalid_rate", "abstain_rate"):
ax.set_title("\n".join(textwrap.wrap(title, width=46)), weight="bold", pad=12, fontsize=11)
else:
ax.set_title(title, weight="bold", pad=12)
ax.set_xlabel("% of training labels polluted")
ax.set_ylabel(PRETTY[metric])
ax.set_xticks(POLLUTION_LEVELS)
ax.yaxis.set_major_formatter(mticker.PercentFormatter(xmax=1.0))
ax.legend(title="method", frameon=True, fontsize=9)
sns.despine()
plt.tight_layout()
out.parent.mkdir(parents=True, exist_ok=True)
plt.savefig(out, dpi=150)
plt.close()
print(f"wrote {out}")
def make_plots():
if not RESULTS_CSV.exists():
raise FileNotFoundError(f"{RESULTS_CSV} not found — run without --plot-only first.")
FIGDIR.mkdir(parents=True, exist_ok=True)
df = pd.read_csv(RESULTS_CSV)
off = df[df["runtime"] == "OFF"]
on = df[df["runtime"] == "ON"]
n_runs = df["run"].nunique()
metrics = [m for m in METRICS if m in df.columns] # tolerate older CSVs missing a metric
# summaries: mean/std over the runs, per (lambda, method, level)
for view, path in ((off, SUMMARY_CSV), (on, RUNTIME_ON_CSV)):
summary = (view.groupby(["lambda", "method", "percentage"])[metrics]
.agg(["mean", "std"]).reset_index())
summary.columns = ["_".join(c).rstrip("_") for c in summary.columns]
summary.to_csv(path, index=False)
print(f"wrote {path}")
# MAIN figures: runtime OFF, one figure per (lambda, metric)
for lam in sorted(df["lambda"].unique()):
sub_lam = off[off["lambda"] == lam]
for metric in metrics:
_lineplot(sub_lam, metric,
f"{PRETTY[metric]} vs. label pollution (λ={lam}, {n_runs} runs, runtime OFF)",
FIGDIR / f"lambda{lam}" / f"{metric}.png")
# EXTRA figures: runtime ON, one figure per (lambda, metric)
for lam in sorted(on["lambda"].unique()):
sub_lam = on[on["lambda"] == lam]
for metric in metrics:
_lineplot(sub_lam, metric,
f"{PRETTY[metric]} vs. label pollution (λ={lam}, {n_runs} runs, runtime ON)",
FIGDIR / "runtime_ON" / f"{metric}_lambda{lam}.png")
# ── pooled out-of-fold metrics + paired guided-vs-unguided tests ──────────────
def _pred_is_invalid(asset, y_pred):
"""True if y_pred is ontologically invalid for `asset` (same notion the runtime guidance
uses). Unknown/empty asset, or an asset with no known valid-action set -> not counted."""
if str(y_pred) == ABSTAIN: # abstention isn't an invalid action (tracked via abstain_rate)
return False
if not isinstance(asset, str) or not asset:
return False
valid = {a.strip().lower() for a in get_ontology_info_actions_per_asset(asset)}
return bool(valid) and str(y_pred).strip().lower() not in valid
def make_pooled_metrics():
"""Collapse each repeat's five out-of-fold prediction sets into ONE pooled score.
Every row is predicted exactly once per repeat (in the fold where it's held out), so a
macro-F1 over the pooled ~584 predictions uses ALL instances of each rare class at once —
far steadier than averaging five tiny per-fold F1s. One row per
(lambda, level, repeat, method, runtime)."""
if not OOF_CSV.exists():
raise FileNotFoundError(f"{OOF_CSV} not found — run the sweep first (without --plot-only).")
oof = pd.read_csv(OOF_CSV)
keys = ["lambda", "percentage", "repeat", "method", "runtime"]
rows = []
for key, g in oof.groupby(keys, sort=False):
yt = g["y_true"].astype(str)
yp = g["y_pred"].astype(str)
inv = np.mean([_pred_is_invalid(a, p) for a, p in zip(g["asset"], g["y_pred"])])
rows.append({**dict(zip(keys, key)),
"test_accuracy": float((yt == yp).mean()),
"test_f1_macro": f1_score(yt, yp, average="macro", zero_division=0),
"test_f1_weighted": f1_score(yt, yp, average="weighted", zero_division=0),
"invalid_rate": float(inv),
"abstain_rate": float((yp == ABSTAIN).mean()),
"n_rows": int(len(g))})
pooled = pd.DataFrame(rows)
pooled.to_csv(POOLED_CSV, index=False)
print(f"wrote {POOLED_CSV} ({len(pooled)} rows; pooled over {oof['fold'].nunique()} folds/repeat)")
return pooled
def _paired_stats(delta):
"""Summary + significance of a paired difference vector (guided - no-guidance)."""
delta = np.asarray(delta, dtype=float)
n = len(delta)
out = {"n": n,
"mean_delta": float(delta.mean()) if n else float("nan"),
"median_delta": float(np.median(delta)) if n else float("nan"),
"wins": int((delta > 0).sum()), "losses": int((delta < 0).sum()),
"ties": int((delta == 0).sum()),
"wilcoxon_p": float("nan"), "ttest_p": float("nan")}
if n >= 6 and np.any(delta != 0):
try:
out["wilcoxon_p"] = float(wilcoxon(delta, zero_method="wilcox").pvalue)
except ValueError:
pass
if n >= 2 and np.std(delta) > 0:
out["ttest_p"] = float(ttest_rel(delta, np.zeros(n)).pvalue)
return out
def make_paired_tests():
"""Paired guided-vs-unguided macro-F1 tests per (lambda, level, runtime).
Both bases exploit the common-random-numbers design (every method shares each fold's
data/pollution/seed, so fold noise cancels in the difference):
per_fold - matched by run across all repeats x folds (more points; folds within
a repeat aren't independent, so treat its p-values as indicative).
per_repeat_pooled - matched by repeat on the pooled OOF macro-F1 (fewer, cleaner points)."""
perfold = pd.read_csv(RESULTS_CSV)
pooled = pd.read_csv(POOLED_CSV) if POOLED_CSV.exists() else make_pooled_metrics()
rows = []
for basis, df, unit in [("per_fold", perfold, "run"),
("per_repeat_pooled", pooled, "repeat")]:
for (lam, lvl, rt), g in df.groupby(["lambda", "percentage", "runtime"]):
base = g[g["method"] == "no guidance"].set_index(unit)["test_f1_macro"]
for m in GUIDED_METHODS:
gm = g[g["method"] == m].set_index(unit)["test_f1_macro"]
common = base.index.intersection(gm.index)
delta = (gm.loc[common] - base.loc[common]).to_numpy()
rows.append({"basis": basis, "lambda": lam, "percentage": lvl,
"runtime": rt, "method": m, **_paired_stats(delta)})
out = pd.DataFrame(rows)
out.to_csv(PAIRED_CSV, index=False)
print(f"wrote {PAIRED_CSV}")
# compact console view: pooled basis, runtime OFF
view = out[(out["basis"] == "per_repeat_pooled") & (out["runtime"] == "OFF")]
if not view.empty:
print("\nPaired guided-vs-unguided macro-F1 (pooled per repeat, runtime OFF; * = Wilcoxon p<0.05):")
print(f"{'lam':>3} {'poll':>4} {'method':>20} {'meanD':>8} {'win/los':>8} {'wilcox_p':>9}")
for _, r in view.sort_values(["lambda", "percentage", "method"]).iterrows():
sig = r["wilcoxon_p"] < 0.05 if r["wilcoxon_p"] == r["wilcoxon_p"] else False
wl = f"{int(r['wins'])}/{int(r['losses'])}"
print(f"{int(r['lambda']):>3} {int(r['percentage']):>4} {r['method']:>20} "
f"{r['mean_delta']:>+8.4f} {wl:>8} {r['wilcoxon_p']:>9.3f}{' *' if sig else ''}")
return out
def make_pooled_plots():
"""Line plots built from the POOLED out-of-fold metrics (steadier than the per-fold ones)."""
if not POOLED_CSV.exists():
make_pooled_metrics()
pooled = pd.read_csv(POOLED_CSV)
n_rep = pooled["repeat"].nunique()
for rt in ("OFF", "ON"):
sub_rt = pooled[pooled["runtime"] == rt]
for lam in sorted(sub_rt["lambda"].unique()):
s = sub_rt[sub_rt["lambda"] == lam]
for metric in ["test_accuracy", "test_f1_macro", "test_f1_weighted", "invalid_rate", "abstain_rate"]:
out = (FIGDIR / "pooled" / f"lambda{lam}" / f"{metric}.png" if rt == "OFF"
else FIGDIR / "pooled" / "runtime_ON" / f"{metric}_lambda{lam}.png")
_lineplot(s, metric,
f"{PRETTY[metric]} (pooled OOF) vs. label pollution "
f"(λ={lam}, {n_rep}×{N_FOLDS}-fold, runtime {rt})", out)
# ── "comparing to prof": custom variants vs Wittgenstein + Weka on the CLEAN data ─
def _scores(y_true, y_pred):
"""Test accuracy + macro & weighted F1 from true/predicted label arrays."""
y_true, y_pred = np.asarray(y_true), np.asarray(y_pred)
return {"test_accuracy": float((y_pred == y_true).mean()),
"test_f1_macro": f1_score(y_true, y_pred, average="macro", zero_division=0),
"test_f1_weighted": f1_score(y_true, y_pred, average="weighted", zero_division=0)}
def wittgenstein_predict(train_df, test_df, seed):
"""One-vs-rest wittgenstein RIPPER; returns test predictions aligned to test_df."""
clf = OneVsRestClassifier(_SklearnRIPPER(random_state=seed))
with contextlib.redirect_stdout(io.StringIO()):
clf.fit(train_df.drop("y", axis=1), train_df["y"])
return clf.predict(test_df.drop("y", axis=1))
def weka_predict(train_df, test_df, seed, all_classes):
"""Weka JRip test-set predictions (or None if weka.jar/java missing or it errors).
Maps Weka's 1-based class index back via all_classes — robust to label quoting/spaces."""
if not os.path.isfile(WEKA_JAR):
return None
tr = te = None
try:
with tempfile.NamedTemporaryFile("w", suffix=".arff", delete=False) as f: tr = f.name
with tempfile.NamedTemporaryFile("w", suffix=".arff", delete=False) as f: te = f.name
_write_arff(train_df, tr, all_classes)
_write_arff(test_df, te, all_classes)
cmd = ["java", "-cp", WEKA_JAR, "weka.classifiers.rules.JRip",
"-S", str(seed), "-t", tr, "-T", te,
"-classifications", "weka.classifiers.evaluation.output.prediction.CSV"]
out = subprocess.run(cmd, capture_output=True, text=True, check=False).stdout
# prediction CSV rows look like: inst#,actual,predicted,error,prob -> "<idx>:<label>"
idxs = re.findall(r"^\s*\d+,[^,]+,(\d+):", out, re.M)
if len(idxs) < len(test_df):
return None
return [all_classes[int(i) - 1] for i in idxs[-len(test_df):]] # test block = last N
finally:
for p in (tr, te):
if p:
try: os.unlink(p)
except OSError: pass
def load_prof_done():
if not PROF_CSV.exists():
return set()
d = pd.read_csv(PROF_CSV)
return {(int(r.run), str(r.method)) for r in d.itertuples()}
def append_prof_rows(rows):
pd.DataFrame(rows, columns=PROF_COLUMNS).to_csv(
PROF_CSV, mode="a", header=not PROF_CSV.exists(), index=False)
def run_prof_comparison():
"""Clean dataset (no pollution), runtime guidance OFF: the 4 custom variants vs the
Wittgenstein and Weka baselines, over the same 10 repeated-CV runs."""
OUTDIR.mkdir(parents=True, exist_ok=True)
X, y, all_classes = load_clean_dummies()
rskf = RepeatedStratifiedKFold(n_splits=N_FOLDS, n_repeats=N_REPEATS, random_state=SPLIT_SEED)
done = load_prof_done()
print(f"\n[prof] {len(done)} (run,method) units already done — skipping those.")
for run, (tri, tei) in enumerate(rskf.split(X, y), start=1):
train_df, test_df = make_fold_dfs(X, y, tri, tei) # clean, untouched
y_true = test_df["y"].values
rows = []
for name, fn, kw in build_methods(PROF_LAMBDA): # custom variants (runtime OFF)
if (run, name) in done:
continue
np.random.seed(run)
rule_set, sv = fn(train_df.copy(), **kw)
res, _ = evaluate_ruleset(rule_set, test_df.copy(), sv, False)
rows.append({"run": run, "method": name, **_scores(res["y"], res["y_pred"])})
if (run, "Wittgenstein") not in done:
yp = wittgenstein_predict(train_df, test_df, run)
rows.append({"run": run, "method": "Wittgenstein", **_scores(y_true, yp)})
if (run, "WEKA") not in done:
yp = weka_predict(train_df, test_df, run, all_classes)
sc = _scores(y_true, yp) if yp is not None else {m: float("nan") for m in PROF_METRICS}
rows.append({"run": run, "method": "WEKA", **sc})
if rows:
append_prof_rows(rows)
for r in rows:
print(f"[prof] run {run:>2} | {r['method']:<20} acc={r['test_accuracy']:.3f} "
f"f1_macro={r['test_f1_macro']:.3f} f1_weighted={r['test_f1_weighted']:.3f}")
print("[prof] comparison done.")
def make_prof_plots():
if not PROF_CSV.exists():
raise FileNotFoundError(f"{PROF_CSV} not found — run the prof comparison first.")
PROF_FIGDIR.mkdir(parents=True, exist_ok=True)
df = pd.read_csv(PROF_CSV)
metrics = [m for m in PROF_METRICS if m in df.columns]
summary = df.groupby("method")[metrics].agg(["mean", "std"]).reset_index()
summary.columns = ["_".join(c).rstrip("_") for c in summary.columns]
summary.to_csv(PROF_SUMMARY_CSV, index=False)
print(f"wrote {PROF_SUMMARY_CSV}")
order = [m for m in PROF_ORDER if m in df["method"].unique()]
n_runs = df["run"].nunique()
sns.set_theme(style="whitegrid", context="talk")
for metric in metrics:
plt.figure(figsize=(10, 6))
ax = sns.barplot(data=df, x="method", y=metric, order=order, errorbar="ci")
ax.set_title(f"{PRETTY[metric]} on clean data — custom vs prof ({n_runs} runs, runtime OFF)",
weight="bold", pad=12)
ax.set_xlabel("")
ax.set_ylabel(PRETTY[metric])
ax.yaxis.set_major_formatter(mticker.PercentFormatter(xmax=1.0))
plt.setp(ax.get_xticklabels(), rotation=30, ha="right") # rotate without re-setting ticks
sns.despine()
plt.tight_layout()
out = PROF_FIGDIR / f"{metric}.png"
plt.savefig(out, dpi=150)
plt.close()
print(f"wrote {out}")
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--plot-only", action="store_true",
help="skip computation; rebuild summaries + figures from the CSVs")
parser.add_argument("--fresh", action="store_true",
help="recompute everything: back up the existing CSV(s) and start over "
"(use after changing the pollution / guidance / model code)")
parser.add_argument("--prof-only", action="store_true",
help="only run/refresh the clean-data 'comparing to prof' experiment + plots")
args = parser.parse_args()
if args.fresh and not args.plot_only:
import shutil
from datetime import datetime
ts = f"{datetime.now():%Y%m%d_%H%M%S}"
to_reset = [PROF_CSV] if args.prof_only else [RESULTS_CSV, OOF_CSV, PROF_CSV]
for csv in to_reset:
if csv.exists():
bak = csv.with_name(f"{csv.stem}_backup_{ts}.csv")
shutil.move(str(csv), str(bak))
print(f"--fresh: moved {csv.name} -> {bak.name}")
for derived in () if args.prof_only else (POOLED_CSV, PAIRED_CSV): # regenerated from scratch
if derived.exists():
derived.unlink()
if args.prof_only:
if not args.plot_only:
run_prof_comparison()
make_prof_plots()
else:
if not args.plot_only:
run_experiments()
# main-sweep summaries, pooled OOF metrics, paired tests, and plots
make_plots()
make_pooled_metrics()
make_paired_tests()
make_pooled_plots()
# prof comparison is optional: needs java/weka.jar + wittgenstein; never block the main outputs
try:
if not args.plot_only:
run_prof_comparison()
make_prof_plots()
except Exception as e:
print(f"[prof] skipped/failed: {e}")