ボートレース予想システムのデータ基盤づくりで、地味に重たい不具合にハマった。

「フライング（F）や K0（欠場系の表記揺れ）など、着順が数字じゃないレコードが混ざると、投入データが列ズレして欠落する」というやつ。

Excel→MySQL投入の段階でズレると、race_id や race_date まで空になるケースが出て、これはさすがに根本対応が必要になった。

症状：K0が来た瞬間に列ズレして、キー列まで欠ける

当初の状態では、以下のような現象が起きていた。

• K0（やF等）が混ざると token 数が変わって列位置がズレる
• ズレが波及し、race_id / race_date / course_code / rno あたりが空白になる
• 暫定対処として「最後の列を削る」などをやると、逆に整合性が壊れて悪化

結論：“固定インデックス前提のsplit”はやめる。これに尽きる。

目標：1行＝1艇を崩さず、MySQLに正しく入れる

今回のゴールはシンプル。

• 「(2) 出走表も含めた “1行＝1艇” データ（展示/進入/ST/タイム）」をMySQLで安定保持
• K0/Fなどが来ても列ズレさせない
• 既存DDL（主キーなど）を維持したまま投入（replaceで壊さない）

設計判断：to_sql(replace)は禁止。append + upsertへ

df.to_sql(if_exists="replace") は一見ラクやけど、これやるとテーブル再作成になるので、

• 主キーが消える
• インデックスが消える
• 型が変わる

みたいな事故が起きる。

なので方針はappend + upsert（ON DUPLICATE KEY UPDATE）に切り替えた。

主キーは (race_id, Lane) にして「1レース1艇1行」を保証。

ハマり1：ArgumentError と orig の罠（原因が見えない）

最初に出たログがこれ。

Connected to MySQL!
Error connecting to MySQL: 'ArgumentError' object has no attribute 'orig'
[NG] upsert failed

「え、接続できてるのに“connecting”で落ちたことになってる？」と思ったら、原因は二段構え。

• 本当のエラーは upsert で発生している
• 例外処理が e.orig を見に行って さらに落ちる（ArgumentErrorにはorigがない）

まずはエラーの足跡を全部出すために、例外処理をこう変えた。

except Exception as e:
    print(f"[NG] upsert failed: {type(e).__name__}: {e}")
    if hasattr(e, "orig"):
        print(f"  orig: {e.orig}")
    traceback.print_exc()
    return False

これで本当の原因が見えるようになった。

ハマり2：pandasのSQLTableをinsert()に渡してた

本体のエラーはこれ。

sqlalchemy.exc.ArgumentError:
subject table for an INSERT, UPDATE or DELETE expected,
got <pandas.io.sql.SQLTable object ...>

つまり、mysql_insert(table) に渡してた table が、SQLAlchemyのTableじゃなくてpandasのSQLTableだった。

修正はシンプルで、pandas側の table.table を使う。

解決：upsert_method を正しく実装（PK自動取得で大小文字ズレも防止）

最終的に落ち着いた upsert_method がこれ。

def upsert_method(table, conn, keys, data_iter):
    sa_table = table.table  # pandas SQLTable -> SQLAlchemy Table

    rows = [dict(zip(keys, row)) for row in data_iter]
    if not rows:
        return 0

    insert_stmt = mysql_insert(sa_table).values(rows)

    # PK列を自動取得（race_id / Lane の大小文字ズレ事故を防ぐ）
    pk_cols = {c.name for c in sa_table.primary_key.columns}

    update_cols = {
        c.name: insert_stmt.inserted[c.name]
        for c in sa_table.columns
        if c.name not in pk_cols
    }

    upsert_stmt = insert_stmt.on_duplicate_key_update(**update_cols)
    result = conn.execute(upsert_stmt)
    return result.rowcount

これで append しつつ、既存行は update できるようになった。

追加要件：Positionは正規化しつつ、元値は Position_raw に残す

ルールはこう。

• '01'..'06' や '1'..'6' は int
• F/K0/S1/欠/失/転/落... は全部 6
• ただし、元値は Position_raw 列に保持したい（後で検証できるように）

ここで重要なのは、正規化する前に Position_raw を退避すること。

def cleanse_df_for_mysql(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()

    # ★ Position_raw を先に退避（正規化前の元値）
    if "Position" in df.columns:
        df["Position_raw"] = df["Position"].astype("string").str.strip()

        pos_num = pd.to_numeric(df["Position_raw"], errors="coerce")
        df["Position"] = pos_num.where(pos_num.between(1, 6), 6).astype("int64")

    # NaN -> None（MySQLにNULLで入れる）
    df = df.where(pd.notna(df), None)
    return df

DB側には Position_raw 列を追加。

ALTER TABLE brace_result
ADD COLUMN Position_raw VARCHAR(16) NULL;

結果：K0でも欠損せず、既存行もupdateできた

最終ログも良い感じ。

[INFO] importing: K240101pre01_race.xlsx
        race_id  Lane  Position  Tenji  Entry  Start   Time
0  202401012401     1         1   6.89    1.0   0.08  109.7
...
Connected to MySQL!
[OK] upsert done

さらにDB側で確認すると、Position_raw に K0 が残り、Tenji/Entry/Start/Time はNULLでも列ズレせずに保持できていた。

そして既存レコードも upsert で update されることを確認。これでチケット#47は完了。

学び：データクレンジングは「最後を切る」じゃなく「崩さずに吸収」

今回の学びはこれ。

• 例外値（K0/F/欠…）は必ず来る。来た時に列ズレさせないのが最優先
• 暫定で「最後を消す」みたいな対処は、後で必ず地獄を見る
• エラーは、まずスタックトレースを出して本体を見える化する
• pandas x SQLAlchemy の境界は、オブジェクト種別（SQLTable vs Table）に注意

次は、取り込んだ「1行=1艇」データのNULL率や例外値頻度を可視化しつつ、予想ロジック側（特徴量設計）に進めていく。