from __future__ import annotations import ast import json import os import unicodedata from pathlib import Path import joblib import numpy as np import pandas as pd from sklearn.ensemble import RandomForestClassifier from sklearn.multiclass import OneVsRestClassifier BLOCKS = ["High", "Medium", "Low"] BLOCK_TO_IDX = {b: i for i, b in enumerate(BLOCKS)} PHASE_TO_BLOCK = { "Build Up against High Block": "High", "Build Up against Medium Block": "Medium", "Build Up against Low Block": "Low", } RANDOM_SEED = 42 FEATURE_COLS = [ "feat_n_pressure_own_half", "feat_avg_pass_options_rival", "feat_top2_avg_x", "feat_team_avg_x", "feat_back4_avg_x", "feat_lowest_def_no_gk_avg_x", "feat_convex_hull_area_no_gk", "feat_team_width_no_gk", "feat_team_depth_no_gk", "feat_n_players_rival_half_avg", "feat_n_pressure_high", "feat_n_pressure_medium", "feat_n_pressure_low", "feat_avg_event_x", "feat_pct_actions_rival", "feat_n_actions_rival", ] DEFAULT_BLOCK_TRAIN_PREPROCESSED = Path( os.environ.get("RACING_BLOCK_TRAIN_PREPROCESSED", "/Users/pagrois/Documents/Racing/preprocessed_SSD_25-26.csv") ) DEFAULT_BLOCK_TRAIN_RAW_DIR = Path( os.environ.get("RACING_BLOCK_TRAIN_RAW_DIR", "/Users/pagrois/Documents/Racing/raw_events") ) DEFAULT_BLOCK_MODEL_PATH = Path( os.environ.get("RACING_BLOCK_MODEL_PATH", "/Users/pagrois/Racing/data/modeling/sequence_block_model.joblib") ) DEFAULT_BLOCK_FEATURE_CACHE = Path( os.environ.get( "RACING_BLOCK_TRAIN_FEATURE_CACHE", "/Users/pagrois/Racing/reports/league_sequence_features_20260505_rawrefresh.pkl", ) ) SHOT_EVENT_TYPES = {"Goal", "MissedShots", "SavedShot", "ShotOnPost", "ChanceMissed"} PASS_EVENT_NAME = "Pass" TIME_BANDS = [ ("t00_15", 0, 15), ("t15_30", 15, 30), ("t30_45", 30, 45), ("t45_60", 45, 60), ("t60_75", 60, 75), ("t75_90", 75, 90), ("t90_plus", 90, np.inf), ] LANE_BINS = [ ("ext_der", 0.0, 21.0), ("int_der", 21.0, 37.0), ("centro", 37.0, 63.0), ("int_izq", 63.0, 79.0), ("ext_izq", 79.0, 100.0), ] BAND_BINS = [ ("creacion", 50.0, 68.0), ("tres_cuartos", 68.0, 83.0), ("area", 83.0, 100.0), ] ZONE_NAMES = [f"{band}__{lane}" for band, _, _ in BAND_BINS for lane, _, _ in LANE_BINS] def safe_parse(val): if pd.isna(val): return None if isinstance(val, (dict, list)): return val txt = str(val) for loader in (json.loads, ast.literal_eval): try: return loader(txt) except Exception: pass return None def norm_text(txt: str | None) -> str: if txt is None: return "" txt = unicodedata.normalize("NFKD", str(txt)).encode("ascii", "ignore").decode("ascii") txt = txt.lower().strip() return " ".join(txt.replace("-", " ").split()) def parse_match_file_name(path: Path) -> dict[str, str | None]: stem = path.stem out = {"fecha_str": None, "home_name": None, "away_name": None} if " - " not in stem or " vs " not in stem: return out fecha_str, teams_part = stem.split(" - ", 1) home_name, away_name = teams_part.split(" vs ", 1) out["fecha_str"] = fecha_str out["home_name"] = home_name out["away_name"] = away_name return out def infer_league_season_from_path(path: Path, input_root: Path) -> tuple[str | None, str | None]: try: rel = path.relative_to(input_root) except ValueError: rel = path parts = list(rel.parts) if len(parts) >= 3: return parts[0], parts[1] if len(parts) >= 2: return input_root.name, parts[0] return None, None def time_band_from_minute(minute: float) -> str: for label, start, end in TIME_BANDS: if start <= minute < end: return label return "t90_plus" def assign_attack_zone(x: float, y: float) -> str | None: if pd.isna(x) or pd.isna(y): return None x = float(x) y = float(y) if x < 50 or x > 100 or y < 0 or y > 100: return None band_name = None lane_name = None for name, x0, x1 in BAND_BINS: if x0 <= x < x1 or (name == "area" and x <= x1): band_name = name break for name, y0, y1 in LANE_BINS: if y0 <= y < y1 or (name == "izq" and y <= y1): lane_name = name break if band_name is None or lane_name is None: return None return f"{band_name}__{lane_name}" def normalize_rows(arr: np.ndarray) -> np.ndarray: arr = np.asarray(arr, dtype=float) row_sum = arr.sum(axis=1, keepdims=True) out = arr.copy() mask = row_sum.squeeze() > 0 out[mask] = out[mask] / row_sum[mask] out[~mask] = 1.0 / arr.shape[1] return out def probs_from_ovr(model, X: np.ndarray) -> np.ndarray: prob_list = model.predict_proba(X) if isinstance(prob_list, list): probs = np.column_stack([p[:, 1] for p in prob_list]) elif isinstance(prob_list, np.ndarray) and prob_list.ndim == 3: probs = np.column_stack([prob_list[i][:, 1] for i in range(prob_list.shape[0])]) else: probs = np.asarray(prob_list) if probs.ndim == 2 and probs.shape[1] == 3: return probs return probs def dominant_label(df: pd.DataFrame) -> np.ndarray: return np.select( [ (df["w_high"] >= df["w_medium"]) & (df["w_high"] >= df["w_low"]), (df["w_medium"] >= df["w_high"]) & (df["w_medium"] >= df["w_low"]), ], ["High", "Medium"], default="Low", ) def apply_manual_rules(df: pd.DataFrame) -> pd.DataFrame: out = df.copy() out["label_before_rules"] = dominant_label(out) low_a = (out["feat_n_pressure_own_half"] <= 0) & (out["n_passes_7opt_rival"] > out["n_passes"] / 2.0) low_b = out["feat_back4_avg_x"] > 50 high_c = (out["n_passes"] > 7) & (out["feat_back4_avg_x"] < 25) high_d = (out["n_passes_own_half"] >= 6) & ( out["feat_n_pressure_own_half"] >= (out["n_passes_own_half"] / 2.0) ) has_low = low_a | low_b has_high = high_c | high_d both = has_low & has_high out["rule_low_a"] = low_a out["rule_low_b"] = low_b out["rule_high_c"] = high_c out["rule_high_d"] = high_d out["rule_both_high_low"] = both label_after = out["label_before_rules"].copy() label_after = np.where(has_high & ~has_low, "High", label_after) label_after = np.where(has_low & ~has_high, "Low", label_after) label_after = np.where(both, "High+Low", label_after) out["label_after_rules"] = label_after out.loc[has_high & ~has_low, ["w_high", "w_medium", "w_low"]] = [1.0, 0.0, 0.0] out.loc[has_low & ~has_high, ["w_high", "w_medium", "w_low"]] = [0.0, 0.0, 1.0] out.loc[both, ["w_high", "w_medium", "w_low"]] = [0.5, 0.0, 0.5] return out def extract_team_points(ev: dict) -> list[tuple[float, float]]: players: dict[str, tuple[float, float]] = {} def add_one(obj): if not isinstance(obj, dict): return pid = obj.get("playerId") or obj.get("id") or obj.get("shirtNumber") x = obj.get("positionX") y = obj.get("positionY") if pid is None or x is None or y is None: return try: players[str(pid)] = (float(x), float(y)) except Exception: return add_one(safe_parse(ev.get("mainActor"))) pass_target = safe_parse(ev.get("passTarget")) if isinstance(pass_target, dict): for obj in pass_target.get("player", []): add_one(obj) elif isinstance(pass_target, list): for obj in pass_target: add_one(obj) pass_option = safe_parse(ev.get("passOption")) if isinstance(pass_option, dict): objs = pass_option.get("player", []) elif isinstance(pass_option, list): objs = pass_option else: objs = [] for obj in objs: add_one(obj) return list(players.values()) def without_goalkeeper(points: list[tuple[float, float]]) -> list[tuple[float, float]]: if len(points) <= 1: return [] points_sorted = sorted(points, key=lambda p: (p[0], p[1])) return points_sorted[1:] def convex_hull_area(points: list[tuple[float, float]]) -> float: if len(points) < 3: return 0.0 pts = sorted(set((float(x), float(y)) for x, y in points)) if len(pts) < 3: return 0.0 def cross(o, a, b): return (a[0] - o[0]) * (b[1] - o[1]) - (a[1] - o[1]) * (b[0] - o[0]) lower = [] for p in pts: while len(lower) >= 2 and cross(lower[-2], lower[-1], p) <= 0: lower.pop() lower.append(p) upper = [] for p in reversed(pts): while len(upper) >= 2 and cross(upper[-2], upper[-1], p) <= 0: upper.pop() upper.append(p) hull = lower[:-1] + upper[:-1] if len(hull) < 3: return 0.0 area = 0.0 for i in range(len(hull)): x1, y1 = hull[i] x2, y2 = hull[(i + 1) % len(hull)] area += x1 * y2 - x2 * y1 return abs(area) / 2.0 def pass_options_rival_half(ev: dict) -> float: pass_option = safe_parse(ev.get("passOption")) if isinstance(pass_option, dict): players = pass_option.get("player", []) elif isinstance(pass_option, list): players = pass_option else: return np.nan count = 0 for obj in players: if not isinstance(obj, dict): continue x = obj.get("positionX") if x is None: continue try: if float(x) > 50: count += 1 except Exception: continue return float(count) def pressure_intensity_counts(ev: dict) -> tuple[int, int, int]: pressure = safe_parse(ev.get("pressure")) high = medium = low = 0 if isinstance(pressure, dict): for obj in pressure.get("player", []): if not isinstance(obj, dict): continue intensity = str(obj.get("intensity", "")).lower() if intensity == "high": high += 1 elif intensity == "medium": medium += 1 elif intensity == "low": low += 1 return high, medium, low def compute_own_half_passes(df_events: pd.DataFrame) -> pd.DataFrame: cols = ["matchId", "sequenceId", "event_name", "x"] missing = [c for c in cols if c not in df_events.columns] if missing: return pd.DataFrame(columns=["matchId", "sequenceId", "n_passes_own_half"]) tmp = df_events[cols].copy() tmp["sequenceId"] = pd.to_numeric(tmp["sequenceId"], errors="coerce") tmp = tmp.dropna(subset=["sequenceId"]).copy() tmp["sequenceId"] = tmp["sequenceId"].astype(int).astype(str) tmp["x"] = pd.to_numeric(tmp["x"], errors="coerce") tmp = tmp[(tmp["event_name"] == PASS_EVENT_NAME) & (tmp["x"] < 50)].copy() if tmp.empty: return pd.DataFrame(columns=["matchId", "sequenceId", "n_passes_own_half"]) return tmp.groupby(["matchId", "sequenceId"], dropna=False).size().reset_index(name="n_passes_own_half") def extract_sequence_features_from_events(df_events: pd.DataFrame) -> pd.DataFrame: raw_cols = ["matchId", "sequenceId", "event_name", "mainActor", "passOption", "passTarget", "pressure", "x"] missing = [c for c in raw_cols if c not in df_events.columns] if missing: raise ValueError(f"Faltan columnas para extraer features de sequence: {missing}") df_ev = df_events[raw_cols].copy() df_ev["sequenceId"] = pd.to_numeric(df_ev["sequenceId"], errors="coerce") df_ev = df_ev.dropna(subset=["sequenceId"]).copy() df_ev["sequenceId"] = df_ev["sequenceId"].astype(int).astype(str) df_ev["x"] = pd.to_numeric(df_ev["x"], errors="coerce") df_ev["is_pass"] = df_ev["event_name"].eq(PASS_EVENT_NAME) feature_rows = [] for (mid, sid), grp in df_ev.groupby(["matchId", "sequenceId"], sort=False): pass_grp = grp[grp["is_pass"]] top2_vals = [] team_avg_vals = [] back4_vals = [] lowest_def_no_gk_vals = [] hull_area_vals = [] team_width_vals = [] team_depth_vals = [] n_players_rival_vals = [] n_opt_rival_vals = [] n_passes_7opt_rival = 0 for ev in pass_grp.to_dict("records"): pts = extract_team_points(ev) xs = sorted([p[0] for p in pts]) if xs: team_avg_vals.append(float(np.mean(xs))) top2_vals.append(float(np.mean(xs[-2:])) if len(xs) >= 2 else float(xs[-1])) back4_vals.append(float(np.mean(xs[:4])) if len(xs) >= 4 else float(np.mean(xs))) if len(xs) >= 2: lowest_def_no_gk_vals.append(float(xs[1])) n_players_rival_vals.append(float(np.sum(np.array(xs) > 50))) pts_no_gk = without_goalkeeper(pts) if pts_no_gk: xs_no_gk = [p[0] for p in pts_no_gk] ys_no_gk = [p[1] for p in pts_no_gk] hull_area_vals.append(convex_hull_area(pts_no_gk)) team_width_vals.append(float(max(ys_no_gk) - min(ys_no_gk))) team_depth_vals.append(float(max(xs_no_gk) - min(xs_no_gk))) n_opt_rival = pass_options_rival_half(ev) if not np.isnan(n_opt_rival): n_opt_rival_vals.append(float(n_opt_rival)) if n_opt_rival > 7: n_passes_7opt_rival += 1 high_p = medium_p = low_p = 0 n_pressure_own_half = 0 n_pressure_rival = 0 for ev in grp.to_dict("records"): x = ev.get("x") has_pressure = pd.notna(ev.get("pressure")) if has_pressure and pd.notna(x): if float(x) < 50: n_pressure_own_half += 1 elif float(x) > 50: n_pressure_rival += 1 h, m, l = pressure_intensity_counts(ev) high_p += h medium_p += m low_p += l x_vals = grp["x"].dropna().astype(float) n_actions_rival = int((x_vals > 50).sum()) feature_rows.append( { "matchId": mid, "sequenceId": sid, "n_passes": int(pass_grp.shape[0]), "feat_top2_avg_x": float(np.mean(top2_vals)) if top2_vals else np.nan, "feat_avg_pass_options_rival": float(np.mean(n_opt_rival_vals)) if n_opt_rival_vals else np.nan, "feat_n_pressure_own_half": float(n_pressure_own_half), "feat_team_avg_x": float(np.mean(team_avg_vals)) if team_avg_vals else np.nan, "feat_back4_avg_x": float(np.mean(back4_vals)) if back4_vals else np.nan, "feat_lowest_def_no_gk_avg_x": float(np.mean(lowest_def_no_gk_vals)) if lowest_def_no_gk_vals else np.nan, "feat_convex_hull_area_no_gk": float(np.mean(hull_area_vals)) if hull_area_vals else np.nan, "feat_team_width_no_gk": float(np.mean(team_width_vals)) if team_width_vals else np.nan, "feat_team_depth_no_gk": float(np.mean(team_depth_vals)) if team_depth_vals else np.nan, "feat_n_players_rival_half_avg": float(np.mean(n_players_rival_vals)) if n_players_rival_vals else np.nan, "feat_n_pressure_high": float(high_p), "feat_n_pressure_medium": float(medium_p), "feat_n_pressure_low": float(low_p), "feat_avg_event_x": float(x_vals.mean()) if len(x_vals) else np.nan, "feat_pct_actions_rival": float((x_vals > 50).mean()) if len(x_vals) else np.nan, "feat_n_actions_rival": float(n_actions_rival), "n_pressure_rival": float(n_pressure_rival), "n_passes_7opt_rival": float(n_passes_7opt_rival), } ) df_feat = pd.DataFrame(feature_rows) own_half = compute_own_half_passes(df_events) if not own_half.empty: df_feat = df_feat.merge(own_half, on=["matchId", "sequenceId"], how="left") if "n_passes_own_half" not in df_feat.columns: df_feat["n_passes_own_half"] = 0.0 df_feat["n_passes_own_half"] = df_feat["n_passes_own_half"].fillna(0.0).astype(float) return df_feat def resolve_blocks(grp: pd.DataFrame) -> list[str]: max_n = grp["n_ev"].max() tied = set(grp.loc[grp["n_ev"] == max_n, "block_raw"]) if len(tied) == 1: return list(tied) valid = set(grp.loc[(grp["n_ev"] >= 3) & (grp["block_raw"].isin(tied)), "block_raw"]) blocks = list(valid) if valid else list(tied) if len(blocks) > 1 and "Medium" in blocks: no_medium = [b for b in blocks if b != "Medium"] if no_medium: blocks = no_medium return blocks def build_sequence_targets_from_phase_label(df_events: pd.DataFrame) -> pd.DataFrame: if "phaseLabel" not in df_events.columns: return pd.DataFrame(columns=["matchId", "sequenceId", "w_high", "w_medium", "w_low", "source"]) tmp = df_events[["matchId", "sequenceId", "phaseLabel"]].copy() tmp["sequenceId"] = pd.to_numeric(tmp["sequenceId"], errors="coerce") tmp = tmp.dropna(subset=["sequenceId"]).copy() tmp["sequenceId"] = tmp["sequenceId"].astype(int).astype(str) tmp["block_raw"] = tmp["phaseLabel"].map(PHASE_TO_BLOCK) tmp = tmp.dropna(subset=["block_raw"]).copy() if tmp.empty: return pd.DataFrame(columns=["matchId", "sequenceId", "w_high", "w_medium", "w_low", "source"]) block_ev = tmp.groupby(["matchId", "sequenceId", "block_raw"]).size().reset_index(name="n_ev") feature_pass_counts = extract_sequence_features_from_events(df_events)[["matchId", "sequenceId", "n_passes"]] pass_count_map = feature_pass_counts.set_index(["matchId", "sequenceId"])["n_passes"].to_dict() rows = [] for (mid, sid), grp in block_ev.groupby(["matchId", "sequenceId"], sort=False): n_passes = int(pass_count_map.get((mid, sid), -1)) if n_passes < 3: continue grp_multi = grp[grp["n_ev"] >= 3] if n_passes >= 7 and len(grp_multi) > 1: total_multi = grp_multi["n_ev"].sum() weights = {b: 0.0 for b in BLOCKS} for row in grp_multi.itertuples(index=False): weights[row.block_raw] = row.n_ev / total_multi else: blocks = resolve_blocks(grp) weights = {b: 0.0 for b in BLOCKS} share = 1.0 / len(blocks) for block in blocks: weights[block] = share rows.append( { "matchId": mid, "sequenceId": sid, "w_high": weights["High"], "w_medium": weights["Medium"], "w_low": weights["Low"], "source": "opta", } ) return pd.DataFrame(rows) def _build_block_training_frame( preprocessed_path: Path = DEFAULT_BLOCK_TRAIN_PREPROCESSED, feature_cache_path: Path = DEFAULT_BLOCK_FEATURE_CACHE, raw_dir: Path = DEFAULT_BLOCK_TRAIN_RAW_DIR, ) -> pd.DataFrame: pre_cols = ["matchId", "sequenceId", "phaseLabel"] df_pre = pd.read_csv(preprocessed_path, usecols=pre_cols, low_memory=False) df_pre["matchId"] = df_pre["matchId"].astype(str) df_pre["sequenceId"] = pd.to_numeric(df_pre["sequenceId"], errors="coerce") df_pre = df_pre.dropna(subset=["sequenceId"]).copy() df_pre["sequenceId"] = df_pre["sequenceId"].astype(int).astype(str) df_pre["block_raw"] = df_pre["phaseLabel"].map(PHASE_TO_BLOCK) if feature_cache_path.exists(): df_feat = pd.read_pickle(feature_cache_path) else: frames = [] for path in sorted(raw_dir.glob("*.xlsx")): try: df_ev = pd.read_excel(path, sheet_name="Eventos", engine="openpyxl") except Exception: continue if "matchId" not in df_ev.columns: continue df_ev["matchId"] = df_ev["matchId"].astype(str) try: frames.append(extract_sequence_features_from_events(df_ev)) except Exception: continue if not frames: raise RuntimeError("No se pudieron construir features de sequence para entrenar el modelo de bloques.") df_feat = pd.concat(frames, ignore_index=True) feature_cache_path.parent.mkdir(parents=True, exist_ok=True) df_feat.to_pickle(feature_cache_path) block_ev = ( df_pre.dropna(subset=["block_raw"]) .groupby(["matchId", "sequenceId", "block_raw"]) .size() .reset_index(name="n_ev") ) pass_count_map = df_feat.set_index(["matchId", "sequenceId"])["n_passes"].to_dict() target_rows = [] for (mid, sid), grp in block_ev.groupby(["matchId", "sequenceId"], sort=False): n_passes = int(pass_count_map.get((mid, sid), -1)) if n_passes < 3: continue grp_multi = grp[grp["n_ev"] >= 3] if n_passes >= 7 and len(grp_multi) > 1: total_multi = grp_multi["n_ev"].sum() weights = {b: 0.0 for b in BLOCKS} for row in grp_multi.itertuples(index=False): weights[row.block_raw] = row.n_ev / total_multi else: blocks = resolve_blocks(grp) weights = {b: 0.0 for b in BLOCKS} share = 1.0 / len(blocks) for block in blocks: weights[block] = share target_rows.append( { "matchId": mid, "sequenceId": sid, "w_high": weights["High"], "w_medium": weights["Medium"], "w_low": weights["Low"], } ) df_target = pd.DataFrame(target_rows) if df_target.empty: raise RuntimeError("No se pudieron construir targets Opta para el modelo de bloques.") df_model = df_feat.merge(df_target, on=["matchId", "sequenceId"], how="inner") df_model = df_model[df_model["n_passes"] >= 3].dropna(subset=FEATURE_COLS).copy() return df_model def fit_or_load_block_model( model_path: Path = DEFAULT_BLOCK_MODEL_PATH, preprocessed_path: Path = DEFAULT_BLOCK_TRAIN_PREPROCESSED, feature_cache_path: Path = DEFAULT_BLOCK_FEATURE_CACHE, raw_dir: Path = DEFAULT_BLOCK_TRAIN_RAW_DIR, force_retrain: bool = False, ): if model_path.exists() and not force_retrain: return joblib.load(model_path) df_model = _build_block_training_frame( preprocessed_path=preprocessed_path, feature_cache_path=feature_cache_path, raw_dir=raw_dir, ) X_train = df_model[FEATURE_COLS].to_numpy(dtype=float) y_train = (df_model[["w_high", "w_medium", "w_low"]].to_numpy(dtype=float) > 0).astype(int) model = OneVsRestClassifier( RandomForestClassifier( n_estimators=400, min_samples_leaf=5, random_state=RANDOM_SEED, n_jobs=1, class_weight="balanced_subsample", ), n_jobs=1, ) model.fit(X_train, y_train) artifact = {"model": model, "feature_cols": FEATURE_COLS} model_path.parent.mkdir(parents=True, exist_ok=True) joblib.dump(artifact, model_path) return artifact def label_sequences_with_block_model(df_events: pd.DataFrame, model_artifact: dict) -> pd.DataFrame: df_feat = extract_sequence_features_from_events(df_events) if df_feat.empty: return pd.DataFrame() seq_targets = build_sequence_targets_from_phase_label(df_events) if not seq_targets.empty: seq_targets = seq_targets.rename( columns={ "w_high": "opta_w_high", "w_medium": "opta_w_medium", "w_low": "opta_w_low", "source": "opta_source", } ) df_all = df_feat.merge(seq_targets, on=["matchId", "sequenceId"], how="left") feature_cols = list(model_artifact["feature_cols"]) model = model_artifact["model"] predict_mask = df_all["n_passes"] >= 3 predict_mask &= df_all[feature_cols].notna().all(axis=1) if predict_mask.any(): probs = probs_from_ovr(model, df_all.loc[predict_mask, feature_cols].to_numpy(dtype=float)) probs = normalize_rows(probs) pred_idx = probs.argmax(axis=1) pred_labels = np.array(BLOCKS, dtype=object)[pred_idx] df_all.loc[predict_mask, "w_high"] = (pred_labels == "High").astype(float) df_all.loc[predict_mask, "w_medium"] = (pred_labels == "Medium").astype(float) df_all.loc[predict_mask, "w_low"] = (pred_labels == "Low").astype(float) df_all.loc[predict_mask, "source"] = "model" for col in ["w_high", "w_medium", "w_low"]: if col not in df_all.columns: df_all[col] = np.nan opta_available = ( df_all.get("opta_w_high", pd.Series(index=df_all.index)).notna() & df_all.get("opta_w_medium", pd.Series(index=df_all.index)).notna() & df_all.get("opta_w_low", pd.Series(index=df_all.index)).notna() ) missing_model = df_all[["w_high", "w_medium", "w_low"]].isna().all(axis=1) fallback_mask = missing_model & opta_available if fallback_mask.any(): df_all.loc[fallback_mask, "w_high"] = df_all.loc[fallback_mask, "opta_w_high"] df_all.loc[fallback_mask, "w_medium"] = df_all.loc[fallback_mask, "opta_w_medium"] df_all.loc[fallback_mask, "w_low"] = df_all.loc[fallback_mask, "opta_w_low"] df_all.loc[fallback_mask, "source"] = "opta_fallback" df_all = df_all.dropna(subset=["w_high", "w_medium", "w_low"], how="all").copy() if df_all.empty: return df_all df_all[["w_high", "w_medium", "w_low"]] = df_all[["w_high", "w_medium", "w_low"]].fillna(0.0) df_all["source"] = df_all["source"].fillna("model") df_all = apply_manual_rules(df_all) return df_all