from __future__ import annotations import argparse import json import shutil from pathlib import Path import numpy as np import pandas as pd from build_attack_block_team_match_features_from_xlsx import build_team_match_features from download_raw_events_by_league import download_league, list_seasons PROJECT_ROOT = Path(__file__).resolve().parents[1] OUTPUT_ROOT = PROJECT_ROOT / "data" / "modeling" BY_LEAGUE_ROOT = OUTPUT_ROOT / "attack_block_by_league" FINAL_DATASET_PATH = OUTPUT_ROOT / "attack_block_multileague_dataset.parquet" TMP_DOWNLOAD_ROOT = PROJECT_ROOT / "tmp" / "attack_block_raw_events" SHORT_WINDOW = 8 RELATIVE_TARGET_FLOOR = 0.005 OPTA_LEAGUES = [ "Spanish Segunda Division", "Spanish La Liga", "Croatia Prva HNL", "Brack Super League", "Polish Ekstraklasa", "Serbian Super Liga", "Norwegian Eliteserien", "Austrian Bundesliga", "Swedish Allsvenskan", "Danish Superligaen", "Italian Serie B", "France Ligue 2", "Liga Profesional Argentina", ] def _slug(text: str) -> str: return ( text.lower() .replace(" ", "_") .replace("/", "_") .replace("-", "_") .replace("__", "_") ) def _rolling_features(df: pd.DataFrame, source_cols: list[str]) -> pd.DataFrame: df = df.sort_values(["league", "season", "teamId", "fecha", "matchId"]).copy() grouped = df.groupby(["league", "season", "teamId"], sort=False) df["n_prior_matches"] = grouped.cumcount() rolling_frames: dict[str, pd.Series] = {} for col in source_cols: shifted = grouped[col].shift(1) rolling_frames[f"short_mean__{col}"] = shifted.groupby( [df["league"], df["season"], df["teamId"]], sort=False ).transform(lambda s: s.rolling(SHORT_WINDOW, min_periods=1).mean()) rolling_frames[f"long_mean__{col}"] = shifted.groupby( [df["league"], df["season"], df["teamId"]], sort=False ).transform(lambda s: s.expanding(min_periods=1).mean()) rolling_frames[f"long_std__{col}"] = shifted.groupby( [df["league"], df["season"], df["teamId"]], sort=False ).transform(lambda s: s.expanding(min_periods=2).std(ddof=0)) return pd.concat([df, pd.DataFrame(rolling_frames, index=df.index)], axis=1) def _add_opponent_histories(df: pd.DataFrame, hist_cols: list[str]) -> pd.DataFrame: opp = df[["matchId", "teamId", "n_prior_matches"] + hist_cols].copy() rename_map = {"teamId": "opponent_team_id", "n_prior_matches": "opp_n_prior_matches"} rename_map.update({col: f"opp__{col}" for col in hist_cols}) opp = opp.rename(columns=rename_map) return df.merge(opp, on=["matchId", "opponent_team_id"], how="left") def _relative_target(actual: pd.Series, baseline: pd.Series) -> pd.Series: denom = baseline.abs().clip(lower=RELATIVE_TARGET_FLOOR) return 100.0 * ((actual / denom) - 1.0) def _build_relative_targets(df: pd.DataFrame) -> pd.DataFrame: out = df.copy() attack_cols = sorted(c for c in out.columns if c.startswith("actual_attack_share__")) pv_cols = sorted(c for c in out.columns if c.startswith("actual_pv_share__")) conceded_cols = sorted(c for c in out.columns if c.startswith("actual_conceded_attack_share__") or c.startswith("actual_conceded_pv_share__")) for col in attack_cols + pv_cols + conceded_cols: baseline_col = f"long_mean__{col}" if baseline_col in out.columns: out[f"target_rel__{col}"] = _relative_target(out[col], out[baseline_col]) if "xg_total" in out.columns and "long_mean__xg_total" in out.columns: out["target_rel__xg_total"] = _relative_target(out["xg_total"], out["long_mean__xg_total"]) return out def _assign_temporal_split(df: pd.DataFrame) -> pd.DataFrame: match_dates = df[["matchId", "fecha"]].drop_duplicates().sort_values(["fecha", "matchId"]).reset_index(drop=True) split_idx = min(max(int(np.floor(len(match_dates) * 0.9)), 1), len(match_dates) - 1) cutoff = match_dates.loc[split_idx, "fecha"] out = df.copy() out["split"] = np.where(out["fecha"] >= cutoff, "test", "train") return out def build_modeling_dataset(df_base: pd.DataFrame) -> tuple[pd.DataFrame, dict]: df = df_base.copy() df["fecha"] = pd.to_datetime(df["fecha"], errors="coerce") meta_cols = { "matchId", "fecha", "league", "season", "teamId", "opponent_team_id", "team_name", "opponent_name", "home_name", "away_name", "is_home", } source_cols = [c for c in df.columns if c not in meta_cols and pd.api.types.is_numeric_dtype(df[c])] df = _rolling_features(df, source_cols=source_cols) hist_cols = sorted(c for c in df.columns if c.startswith("short_mean__") or c.startswith("long_mean__") or c.startswith("long_std__")) df = _add_opponent_histories(df, hist_cols) df = _build_relative_targets(df) df = _assign_temporal_split(df) target_cols = [c for c in df.columns if c.startswith("target_rel__")] df["usable_for_model"] = ( df["n_prior_matches"].fillna(0) >= 1 ) & ( df["opp_n_prior_matches"].fillna(0) >= 1 ) & ( df[target_cols].notna().any(axis=1) ) summary = { "rows_total": int(len(df)), "matches_total": int(df["matchId"].nunique()), "leagues": sorted(df["league"].dropna().astype(str).unique().tolist()), "seasons": sorted(df["season"].dropna().astype(str).unique().tolist()), "feature_columns": int(sum(c.startswith("short_mean__") or c.startswith("long_mean__") or c.startswith("long_std__") or c.startswith("opp__") for c in df.columns)), "target_columns": int(len(target_cols)), "usable_rows": int(df["usable_for_model"].sum()), "train_rows": int((df["split"] == "train").sum()), "test_rows": int((df["split"] == "test").sum()), "target_xg_available_rows": int(df["target_rel__xg_total"].notna().sum()) if "target_rel__xg_total" in df.columns else 0, "relative_target_floor": RELATIVE_TARGET_FLOOR, } return df, summary def build_all_leagues( leagues: list[str], tmp_download_root: Path, output_root: Path, final_dataset_path: Path, cleanup_raw: bool, overwrite_downloads: bool, force_retrain_block_model: bool, existing_root: Path | None, score_state: str, ) -> tuple[pd.DataFrame, dict]: output_root.mkdir(parents=True, exist_ok=True) by_league_root = output_root / "attack_block_by_league" by_league_root.mkdir(parents=True, exist_ok=True) league_summaries = [] league_parquets = [] for idx, league in enumerate(leagues, start=1): print(f"\n=== [{idx}/{len(leagues)}] {league} ===", flush=True) if existing_root is not None: league_input = existing_root / league seasons = sorted([p.name for p in league_input.iterdir() if p.is_dir()]) if league_input.exists() else [] dl_summary = { "league": league, "seasons": seasons, "files_total": None, "downloaded": 0, "skipped": 0, "failed": 0, "local_root": str(existing_root), "mode": "existing_local", } else: local_root = tmp_download_root seasons = list_seasons(league) dl_summary = download_league( league=league, local_root=local_root, seasons=seasons, overwrite=overwrite_downloads, ) league_input = local_root / league suffix = "" if score_state == "all" else f"__{score_state.replace('-', '_')}" league_output = by_league_root / f"{_slug(league)}{suffix}.parquet" _, feat_summary = build_team_match_features( input_dir=league_input, output_path=league_output, force_retrain_block_model=force_retrain_block_model and idx == 1, score_state=score_state, ) league_summaries.append({"download": dl_summary, "features": feat_summary}) league_parquets.append(league_output) if cleanup_raw and existing_root is None and league_input.exists(): shutil.rmtree(league_input) frames = [pd.read_parquet(path) for path in league_parquets if path.exists()] if not frames: raise RuntimeError("No se generaron parquets por liga.") df_base = pd.concat(frames, ignore_index=True) df_model, summary = build_modeling_dataset(df_base) final_dataset_path.parent.mkdir(parents=True, exist_ok=True) df_model.to_parquet(final_dataset_path, index=False) summary["per_league"] = league_summaries summary["score_state"] = score_state summary_path = final_dataset_path.with_name(final_dataset_path.stem + "_summary.json") summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8") return df_model, summary def main() -> None: parser = argparse.ArgumentParser(description="Descarga ligas, resume XLSX y arma dataset multiliga causal.") parser.add_argument("--league", action="append", dest="leagues") parser.add_argument("--tmp-download-root", type=Path, default=TMP_DOWNLOAD_ROOT) parser.add_argument("--output-root", type=Path, default=OUTPUT_ROOT) parser.add_argument("--final-dataset-path", type=Path, default=FINAL_DATASET_PATH) parser.add_argument("--no-cleanup-raw", action="store_true") parser.add_argument("--overwrite-downloads", action="store_true") parser.add_argument("--force-retrain-block-model", action="store_true") parser.add_argument("--existing-root", type=Path) parser.add_argument("--score-state", choices=["all", "0-0"], default="all") args = parser.parse_args() leagues = args.leagues or OPTA_LEAGUES _, summary = build_all_leagues( leagues=leagues, tmp_download_root=args.tmp_download_root, output_root=args.output_root, final_dataset_path=args.final_dataset_path, cleanup_raw=not args.no_cleanup_raw, overwrite_downloads=args.overwrite_downloads, force_retrain_block_model=args.force_retrain_block_model, existing_root=args.existing_root, score_state=args.score_state, ) print(json.dumps(summary, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()