etl_full.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import re\n",
    "\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "import pandera as pa\n",
    "\n",
    "INPUT_PATH = \"./data/full_classes.csv\"\n",
    "OUTPUT_PATH = \"./data/full_classes.parquet\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "Kats = [\n",
    "    \"offene Klasse\",\n",
    "    \"M Offene Klasse\",\n",
    "    \"W Offene Klasse\",\n",
    "    \"M Ü60\",\n",
    "    \"Ü30\",\n",
    "    \"Ü35\",\n",
    "    \"Ü40\",\n",
    "    \"Ü45\",\n",
    "    \"Ü50\",\n",
    "    \"Ü55\",\n",
    "    \"Ü60\",\n",
    "]  # Example list of category names\n",
    "\n",
    "schema = pa.DataFrameSchema(\n",
    "    {\n",
    "        \"Pos\": pa.Column(pa.Int, checks=pa.Check.greater_than_or_equal_to(0)),\n",
    "        \"Nr\": pa.Column(pa.Int, checks=[pa.Check.greater_than_or_equal_to(0)]),\n",
    "        \"Name\": pa.Column(\n",
    "            pa.String,\n",
    "            checks=pa.Check(\n",
    "                lambda s: not s.isnumeric(),\n",
    "                element_wise=True,\n",
    "                error=\"Name is numeric\",\n",
    "            ),\n",
    "        ),\n",
    "        \"Zeit\": pa.Column(\n",
    "            pa.String,\n",
    "            checks=pa.Check(\n",
    "                lambda x: re.match(r\"\\d{2}:\\d{2}:\\d{2}\\.\\d\", x) is not None,\n",
    "                element_wise=True,\n",
    "            ),\n",
    "        ),\n",
    "        \"Kat\": pa.Column(pa.String, checks=pa.Check.isin(Kats)),\n",
    "        \"KPos\": pa.Column(pa.Int, checks=pa.Check.greater_than_or_equal_to(0)),\n",
    "        \"GPos\": pa.Column(pa.Int, checks=pa.Check.greater_than_or_equal_to(0)),\n",
    "        \"Nation\": pa.Column(\n",
    "            pa.String, checks=pa.Check.str_length(min_value=3, max_value=3)\n",
    "        ),\n",
    "        \"Verein\": pa.Column(pa.String),\n",
    "    },\n",
    "    unique=[\"Nr\"],\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "raw_df = pd.read_csv(INPUT_PATH, header=None, names=range(9))\n",
    "raw_df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "col_names = [\"Pos\", \"Nr\", \"Name\", \"Zeit\", \"Kat\", \"KPos\", \"GPos\", \"Nation\", \"Verein\"]\n",
    "\n",
    "\n",
    "def label_columns(df):\n",
    "    df.columns = col_names\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def clean_dashes_to_nans(df):\n",
    "    df = df.replace(\"-\", np.NaN)\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "time_pattern = r\"\\d{2}:\\d{2}:\\d{2}.\\d\"\n",
    "\n",
    "\n",
    "def clean_squished_name_time(df):\n",
    "    # Split the column at index 2 into name and time using rsplit()\n",
    "    name_time = df.iloc[:, 2].str.rsplit(\" \", n=1, expand=True)\n",
    "    # Create a boolean mask to filter rows where the time matches the pattern\n",
    "    mask = name_time[1].str.match(time_pattern, na=False)\n",
    "    df.loc[mask, \"Name\"] = name_time[0]\n",
    "    # Insert a new column to the right with the extracted times for matching rows\n",
    "    df.loc[mask, \"Zeit\"] = name_time[1]\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def clean_split_name_time_namecol(df):\n",
    "    mask = df[\"Name\"].str.contains(time_pattern, regex=True, na=False)\n",
    "    df.loc[mask, \"Zeit\"] = df.loc[mask, \"Name\"].str.extract(\n",
    "        rf\"({time_pattern})\", expand=False\n",
    "    )\n",
    "    df.loc[mask, \"Name\"] = df.loc[mask, \"Name\"].str.replace(\n",
    "        time_pattern, \"\", regex=True\n",
    "    )\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def clean_time_in_nr(df):\n",
    "    mask = df[\"Nr\"].str.contains(time_pattern, regex=True, na=False)\n",
    "    replacements = [\n",
    "        (\"Verein\", \"GPos\"),\n",
    "        (\"Nation\", \"KPos\"),\n",
    "        (\"GPos\", \"Kat\"),\n",
    "        (\"KPos\", \"Zeit\"),\n",
    "        (\"Kat\", \"Name\"),\n",
    "        (\"Zeit\", \"Nr\"),\n",
    "    ]\n",
    "    for col1, col2 in replacements:\n",
    "        df.loc[mask, col1] = df.loc[mask, col2]\n",
    "\n",
    "    df.loc[mask, \"Nr\"] = df.loc[mask, \"Pos\"].str.split(\" \", n=2).str[1].str.strip()\n",
    "    df.loc[mask, \"Name\"] = (\n",
    "        df.loc[mask, \"Pos\"].str.split(\" \", n=2).str[2:].str.join(\"\").str.strip()\n",
    "    )\n",
    "    df.loc[mask, \"Pos\"] = df.loc[mask, \"Pos\"].str.split(\" \", n=1).str[0].str.strip()\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def clean_drop_nans_and_label_rows(df):\n",
    "    df = df.drop_duplicates()\n",
    "    df = df.dropna(thresh=len(df.columns) - 5)\n",
    "    df = df[~df.apply(lambda row: row.astype(str).str.contains(\"GPos\").any(), axis=1)]\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def clean_move_nation_to_verein(df):\n",
    "    mask = (~df[\"Nation\"].isna()) & (df[\"Verein\"].isna())\n",
    "    df.loc[mask, \"Verein\"] = df.loc[mask, \"Nation\"]\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def clean_convert_pos(df):\n",
    "    df[\"Pos\"] = df[\"Pos\"].str.split(\".\").str[0].str.strip()\n",
    "    mask = df[\"Pos\"] == \"DNF\"\n",
    "    df.loc[mask, \"Zeit\"] = \"00:00:00.0\"\n",
    "    df.loc[mask, \"GPos\"] = 0\n",
    "    df.loc[mask, \"KPos\"] = 0\n",
    "    df.loc[mask, \"Pos\"] = 0\n",
    "    df[\"Pos\"] = df[\"Pos\"].astype(int)\n",
    "    df[\"GPos\"] = df[\"GPos\"].astype(int)\n",
    "    df[\"KPos\"] = df[\"KPos\"].astype(int)\n",
    "\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def clean_convert_nr(df):\n",
    "    df[\"Nr\"] = df[\"Nr\"].astype(int)\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def clean_dedash_name(df):\n",
    "    df[\"Name\"] = df[\"Name\"].apply(\n",
    "        lambda text: text[:-1].strip() if text.endswith(\"-\") else text\n",
    "    )\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def clean_spliced_names_times(df):\n",
    "    mask = (~(df[\"Pos\"] == 0)) & (df[\"Zeit\"].isna())\n",
    "    df.loc[mask, \"Zeit\"] = df.loc[mask, \"Name\"].apply(\n",
    "        lambda s: \"\".join([c for c in s if (c.isnumeric() or c == \":\" or c == \".\")])\n",
    "    )\n",
    "    df.loc[mask, \"Name\"] = df.loc[mask, \"Name\"].apply(\n",
    "        lambda s: \"\".join([c for c in s if not (c.isnumeric() or c == \":\" or c == \".\")])\n",
    "    )\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def clean_empty_nation(df):\n",
    "    mask = (df[\"Nation\"] == df[\"Verein\"]) | (df[\"Nation\"].isna())\n",
    "    df.loc[mask, \"Nation\"] = \"GER\"\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def clean_verein_na(df):\n",
    "    df.loc[df[\"Verein\"].isna(), \"Verein\"] = \"None\"\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "raw_df.sample(5, random_state=42)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Cleaning initial data structure"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "clean_df = (\n",
    "    raw_df.pipe(label_columns)\n",
    "    .pipe(clean_dashes_to_nans)\n",
    "    .pipe(clean_squished_name_time)\n",
    "    .pipe(clean_split_name_time_namecol)\n",
    "    .pipe(clean_time_in_nr)\n",
    "    .pipe(clean_drop_nans_and_label_rows)\n",
    "    .pipe(clean_move_nation_to_verein)\n",
    "    .pipe(clean_convert_pos)\n",
    "    .pipe(clean_convert_nr)\n",
    "    .pipe(clean_dedash_name)\n",
    "    .pipe(clean_spliced_names_times)\n",
    "    .pipe(clean_empty_nation)\n",
    "    .pipe(clean_verein_na)\n",
    ")\n",
    "clean_df.reset_index(drop=True, inplace=True)\n",
    "clean_df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "clean_df.sample(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "clean_df.info()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "clean_df.describe(include=\"all\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Validation and Probing"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "schema.validate(clean_df)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Featurizing for Smarter Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "Geschlechter = [\"M\", \"W\", \"U\"]  # Männlich, weiblich, and unknown/unbekannt\n",
    "\n",
    "new_columns = {\n",
    "    \"DNF\": pa.Column(pa.Bool),\n",
    "    \"NameLen\": pa.Column(pa.Int, pa.Check.greater_than_or_equal_to(0)),\n",
    "    \"FirstLetterRank\": pa.Column(pa.Int, pa.Check.between(0, 26)),\n",
    "    \"VName\": pa.Column(\n",
    "        pa.String,\n",
    "        checks=pa.Check(\n",
    "            lambda s: not s.isnumeric(),\n",
    "            element_wise=True,\n",
    "            error=\"Name is numeric\",\n",
    "        ),\n",
    "    ),\n",
    "    \"FName\": pa.Column(\n",
    "        pa.String,\n",
    "        checks=pa.Check(\n",
    "            lambda s: not s.isnumeric(),\n",
    "            element_wise=True,\n",
    "            error=\"Name is numeric\",\n",
    "        ),\n",
    "    ),\n",
    "    \"NoName\": pa.Column(pa.Bool),\n",
    "    \"NoFName\": pa.Column(pa.Bool),\n",
    "    \"MTeam\": pa.Column(pa.Bool),\n",
    "    \"Geschlecht\": pa.Column(pa.String, pa.Check.isin(Geschlechter)),\n",
    "}\n",
    "\n",
    "# Add the new columns to the existing schema\n",
    "full_schema = schema.add_columns(new_columns)\n",
    "full_schema = full_schema.update_column(\n",
    "    \"Zeit\", dtype=pa.Float, checks=[pa.Check.between(0, 12 * 3600)]\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def convert_time_from_str(df):\n",
    "    df[\"Zeit\"] = pd.to_timedelta(df[\"Zeit\"]).dt.total_seconds()\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "letter_rank = {\n",
    "    letter: rank for rank, letter in enumerate(\"abcdefghijklmnopqrstuvwxyz\", start=1)\n",
    "}\n",
    "\n",
    "# Function to get the rank of the first letter in a name\n",
    "\n",
    "\n",
    "def get_first_letter_rank(name):\n",
    "    first_letter = name[0].lower()\n",
    "    return letter_rank.get(first_letter, 0)\n",
    "\n",
    "\n",
    "def featurize(df):\n",
    "    df[\"DNF\"] = df[\"Pos\"] == 0\n",
    "    df[\"NameLen\"] = df[\"Name\"].str.len()\n",
    "    df[\"FirstLetterRank\"] = df[\"Name\"].apply(get_first_letter_rank)\n",
    "    df[\"VName\"] = df[\"Name\"].str.split().str[:-1].apply(\" \".join).str.strip()\n",
    "    df[\"FName\"] = df[\"Name\"].str.split().str[-1].str.strip()\n",
    "    df[\"NoName\"] = (df[\"VName\"] == \"No\") & (df[\"FName\"] == \"Name\")\n",
    "    df[\"NoFName\"] = (df[\"FName\"] == \"Noname\") | (df[\"NoName\"])\n",
    "    df[\"MTeam\"] = df[\"Verein\"] == \"MERENTIS GmbH\"\n",
    "    df[\"MTeam\"] = df[\"MTeam\"].astype(bool)\n",
    "    df = df.sort_values(by=[\"Kat\", \"Zeit\"]).reset_index(drop=True)\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "full_df = clean_df.pipe(convert_time_from_str).pipe(featurize)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "men_df = pd.read_csv(\"./data/classes_m.csv\", header=None, names=range(9))\n",
    "men_df = (\n",
    "    men_df.pipe(label_columns)\n",
    "    .pipe(clean_dashes_to_nans)\n",
    "    .pipe(clean_squished_name_time)\n",
    "    .pipe(clean_split_name_time_namecol)\n",
    "    .pipe(clean_time_in_nr)\n",
    "    .pipe(clean_drop_nans_and_label_rows)\n",
    "    .pipe(clean_convert_nr)\n",
    ")\n",
    "men_nr = men_df[\"Nr\"]\n",
    "\n",
    "\n",
    "women_df = pd.read_csv(\"./data/classes_w.csv\", header=None, names=range(9))\n",
    "women_df = (\n",
    "    women_df.pipe(label_columns)\n",
    "    .pipe(clean_dashes_to_nans)\n",
    "    # .pipe(clean_squished_name_time)\n",
    "    .pipe(clean_split_name_time_namecol)\n",
    "    .pipe(clean_time_in_nr)\n",
    "    .pipe(clean_drop_nans_and_label_rows)\n",
    "    .pipe(clean_convert_nr)\n",
    ")\n",
    "women_nr = women_df[\"Nr\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "conditions = [full_df[\"Nr\"].isin(men_df[\"Nr\"]), full_df[\"Nr\"].isin(women_df[\"Nr\"])]\n",
    "\n",
    "choices = [\"M\", \"W\"]\n",
    "\n",
    "full_df[\"Geschlecht\"] = np.select(conditions, choices, default=\"U\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "full_schema.validate(full_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "full_df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "full_df.describe(include=\"all\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "full_df.to_parquet(OUTPUT_PATH)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "full_df[\n",
    "    (full_df[\"Geschlecht\"] == \"U\")\n",
    "    & (~full_df[\"Verein\"].str.contains(\"Fruitful\"))\n",
    "    & (~full_df[\"DNF\"])\n",
    "]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ok_df = (\n",
    "    full_df[(full_df[\"Kat\"] == \"offene Klasse\") & (full_df[\"Geschlecht\"] == \"W\")]\n",
    "    .sort_values(by=\"GPos\", ascending=True)\n",
    "    .reset_index(drop=True)\n",
    ")\n",
    "jumps_ok = ok_df[ok_df[\"Zeit\"] < ok_df[\"Zeit\"].shift(1)]\n",
    "jumps_ok"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ok_df[(ok_df[\"GPos\"].isin(jumps_ok[\"GPos\"] - 1))].sort_values(\"GPos\")"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}