vendor/scripts/build_multileague_attack_block_dataset.py

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()