【ダッシュボード】データ自動取得（電力価格）

この記事は「【リアルタイム電力予測】需要・価格・電源最適化ダッシュボード構築記」シリーズの20日目です
Day1はこちら | 全体構成を見る

今日は、ダッシュボードの土台となる「毎日自動で更新される入力データ」 のうち「電力価格データ」を取得します。※前回は為替と燃料価格を取得しました
電力価格とは何かについては【電力価格予測】データ構造と取得は、特徴については、【電力価格予測】EDA（価格、燃料・為替、日付）でも記載しています。

データ取得元

JEPX 市場価格

今までのデータ取得方法と異なる点は、ページの中で項目をクリックしていきデータを取得しなければいけないことです。つまり、クリックの動作が複数ある動きをコードで表さなければいけません。

選択項目

選択項目に分岐が多いように見えましたが、それはグラフで可視化したいときのみで、データをダウンロードするだけならどの項目を選択していても全期間（年始～ダウンロード日）の全データがダウンロードされることが分かりました。
つまり、以下の操作だけでよいことになります。

• 表示項目：「約定価格 入札・約定量」 ※操作必要なし
• 値切替：「30分コマ」※操作必要なし
• 受渡日：「日付を選択してください。」→ カレンダーのどこでも良いので選択 ※データダウンロードに日付は関係ない
• 表示期間：「日」※操作必要なし
• 約定価格：デフォルトの「システムプライス」※操作必要なし
• 入札・約定量：デフォルトの「約定総量」※操作必要なし
• 「データダウンロード」：「2025年度」→ データダウンロード

⇒このように選択していくと、2025年度の全項目のデータがダウンロードされるので後は必要に応じてカラムを絞るだけです。
つまり、まとめるとダウンロードに際して必要な動作は以下になります！

1. ブラウザ起動（headless=Trueで画面非表示）
2. ページを開く
3. 「受渡日」入力欄をクリックしてカレンダーを表示 → 今日セルをクリック
4. 「データダウンロード」ボタン → モーダル表示 → 実行ボタン → CSVダウンロードを待つ

Playwright

今回はスクレイピング（ブラウザ自動操作）にPlaywrightを使用しました。JEPXのページは、URLを叩くとCSVが直接返る形式ではなく、ページ内のボタンやモーダル（ポップアップ）操作を経て「ダウンロード」が発生します。そのため、ブラウザ操作そのものを自動化できる Playwright を使います。

簡単な文法

1) sync_playwright() と with（同期API + 後始末）

with sync_playwright() as p:
    ...

• Playwrightのリソース（内部プロセス）を with ブロック終了時に自動で片付ける
• 例外が起きてもリークしにくい

2) chromium.launch(headless=True)（ヘッドレス起動）

browser = p.chromium.launch(headless=True)

• 画面を出さずにブラウザを動かす
• クリックやダウンロードなどの「実際のブラウザ挙動」を再現できる

3) new_context(accept_downloads=True, locale=..., timezone_id=...)（実行環境を固定）

ctx = browser.new_context(
    accept_downloads=True,
    locale="ja-JP",
    timezone_id="Asia/Tokyo"
)

• accept_downloads=True：ダウンロード機能を有効化（これがないと保存できないことがある）
• locale：日本語UI前提のテキスト検索（get_by_text）が安定する
• timezone_id：日付が絡むUI（カレンダー等）の挙動がズレにくい

4) page.goto(..., wait_until="domcontentloaded")（ページを開く）

page.goto(URL, wait_until="domcontentloaded")

• DOMが構築されるタイミングまで待つ
• ただし「ボタンが後から描画される」サイトもあるので、6)の wait_for と併用する

5) Locator を使う（要素の指定：get_by_text / locator）

page.get_by_text("日付を選択してください。").first
page.locator(".ui-datepicker")
page.locator("#modal-box--spot_summary")

• セレクタ（CSS）や表示テキストで要素を取れる
• .first を付けるのは「同じテキストが複数ある」場合の曖昧さ回避
• modal = page.locator(...) のように 範囲（スコープ）を切る と誤クリックが減る

6) wait_for(state="visible")（クリック前に“表示待ち”する）

modal.wait_for(state="visible", timeout=10_000)

• Playwrightでのタイムアウトの多くは「要素がまだ表示されていないのにクリックした」ことが原因
• クリック前に visible を待つ

7) XPath を使ってカレンダーの“今日セル”を特定する

day = panel.locator(f"xpath={xpath}")
day.first.click(force=True)

• カレンダーは「前月/翌月のセル」「無効セル」が混ざりやすい
• XPathで other/disabled を除外し、かつ表示文字（今日の日付）で一点に絞っている
• force=True は「見えているけど上に薄い要素が被っている」などのケースに対応

8) page.expect_download()（クリック→ダウンロードを同期）

with page.expect_download(timeout=120_000) as dlinfo:
    modal.locator('button.dl-button[type="submit"]').click()
download = dlinfo.value
download.save_as(RAW_OUT)

• ダウンロードはHTTPレスポンスではなく **ブラウザの“ダウンロードイベント”**として発生する
• expect_download により「クリックしたのにファイルを取り逃す」を防げる
• CIでは遅いことがあるので timeout=120_000 のように長めに設定

9) tracing.start / tracing.stop と screenshot（失敗時の調査）

page.context.tracing.start(...)
page.screenshot(path=..., full_page=True)
page.context.tracing.stop(path=...)

• 失敗した時に「画面（screenshot）」と「操作ログ（trace.zip）」が残る
• GitHub Actionsのように画面が見えない環境で、原因究明の最短ルート

事前の設定

cacheの保存先等を記載しておきます。

from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeoutError

JST = timezone(timedelta(hours=9))
CACHE_DIR = Path("data/cache")
RAW_DIR = Path("data/raw")
DEBUG_DIR = Path("data/debug")

RAW_OUT  = RAW_DIR / "spot_tokyo_year.csv"
PROC_OUT = CACHE_DIR / "spot_tokyo_bf1w_tdy.parquet"

URL = "https://www.jepx.jp/electricpower/market-data/spot/"

def today_jst_date():
    return datetime.now(JST).date()

カレンダーを開いてセルをクリック

どのセルを教えても一緒ではありますが、ここでは「今日」の日付のセルをクリックするようにしています。

def click_today_cell_only(page):
    """
    - カレンダーを開いて『今日セル』だけをクリック
    - 年/月はデフォルトで当月になっているので触らない
    """
    # トレース & スクショ
    page.context.tracing.start(screenshots=True, snapshots=True, sources=True)

    y = today_jst_date()
    dd = y.day

    # プレースホルダ『日付を選択してください。』をクリック → パネル表示待ち
    page.get_by_text("日付を選択してください。").first.click(timeout=10_000)
    panel = page.locator(".ui-datepicker")
    panel.first.wait_for(state="visible", timeout=10_000) # 可視化を待つ

    # 当月かつ有効な日セルの a/button を XPath で取得 → クリック
    xpath = (
        ".//td[not(contains(@class,'other')) and " # 前後月や無効セルを除外
        "     not(contains(@class,'disabled'))]"
        "//*[self::a or self::button or self::div]"
        f"[normalize-space(text())='{dd}']"
    )
    day = panel.locator(f"xpath={xpath}")

    try:
        day.first.wait_for(state="visible", timeout=5_000)
        day.first.click(timeout=10_000, force=True)
    except PlaywrightTimeoutError as e:
        # デバッグ出力して落とす
        page.screenshot(path=os.path.join(DEBUG_DIR, "calendar_page.png"), full_page=True)
        try: panel.first.screenshot(path=os.path.join(DEBUG_DIR, "calendar_panel.png"))
        except: pass
        page.context.tracing.stop(os.path.join(DEBUG_DIR, "trace.zip"))
        raise RuntimeError(
            "日セルをクリックできませんでした。debug: debug/calendar_page.png / calendar_panel.png / trace.zip"
        ) from e

実行ボタンを押す

# 「データダウンロード」をクリック → 年度選択ポップアップ → 下部のダウンロードをクリック
print("Push data download")
# まずページ本体の “開くボタン” をクリック（モーダルを開く）
page.locator('#filter-section--type button.dl-button[data-dl="spot_summary"]').click(timeout=10_000)

# モーダル表示を待つ
modal = page.locator("#modal-box--spot_summary")
modal.wait_for(state="visible", timeout=10_000)

# モーダル内の “実行ボタン” をクリックして download を待つ（これでCSVが落ちる）
with page.expect_download(timeout=120_000) as dlinfo:
    # name指定だと再び曖昧なので、モーダルを scope にして type=submit を叩く
    modal.locator('button.dl-button[type="submit"]').click()
download = dlinfo.value
download.save_as(RAW_OUT)
print(f"[OK] saved: {RAW_OUT}")

全体の関数

def fetch_price():

    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        ctx = browser.new_context(accept_downloads=True, locale="ja-JP", timezone_id="Asia/Tokyo")
        page = ctx.new_page()
        page.goto("https://www.jepx.jp/electricpower/market-data/spot/", wait_until="domcontentloaded")

        # 日付の選択
        print("Push today cell")
        click_today_cell_only(page)

        # 「データダウンロード」をクリック → 年度選択ポップアップ → 下部のダウンロードをクリック
        print("Push data download")
        # まずページ本体の “開くボタン” をクリック（モーダルを開く）
        page.locator('#filter-section--type button.dl-button[data-dl="spot_summary"]').click(timeout=10_000)

        # モーダル表示を待つ
        modal = page.locator("#modal-box--spot_summary")
        modal.wait_for(state="visible", timeout=10_000)

        # モーダル内の “実行ボタン” をクリックして download を待つ（これでCSVが落ちる）
        with page.expect_download(timeout=120_000) as dlinfo:
            modal.locator('button.dl-button[type="submit"]').click()
        download = dlinfo.value
        download.save_as(RAW_OUT)
        print(f"[OK] saved: {RAW_OUT}")
        
        # 成功トレースも見たいときは保存
        page.context.tracing.stop(path=os.path.join(DEBUG_DIR, "trace_success.zip"))

    df = None
    df = pd.read_csv(filepath_or_buffer=RAW_OUT, encoding="cp932")

    # 列名
    col_date = next(c for c in df.columns if "受渡日" in c)
    col_time = next(c for c in df.columns if "時刻コード" in c or "時間帯" in c)
    col_qty  = next(c for c in df.columns if "約定総量" in c)
    col_offer  = next(c for c in df.columns if "売り入札量" in c)
    col_bid  = next(c for c in df.columns if "買い入札量" in c)
    col_tokyo= next(c for c in df.columns if "エリアプライス東京" in c or ("東京" in c and "プライス" in c))

    out = df.loc[:, [col_date, col_time, col_offer, col_bid, col_qty, col_tokyo]].copy()
    out.columns = ["date", "period", "offer_volume", "bid_volume", "traded_volume", "tokyo_price_jpy_per_kwh"]
    out["date"] = pd.to_datetime(out["date"])

    # 過去1週間分だけ切り出して保存
    today = pd.Timestamp(datetime.now(JST).date())
    before_1w = today - timedelta(days=7)
    end_inclusive = today + pd.Timedelta(hours=23, minutes=30)
    print(f"[FILTER] {before_1w} to {end_inclusive}")
    
    final_out = out.sort_values(["date", "period"]).reset_index(drop=True)
    final_out.rename(columns={"date": "timestamp"}, inplace=True)
    final_out["timestamp"] = (
        final_out["timestamp"]
        + (final_out["period"] - 1).apply(lambda k: pd.Timedelta(minutes=30 * k))
    )
    final_out.drop(columns="period", inplace=True)
    print("ダウンロード結果: \n", final_out)

    final_out = final_out[final_out["timestamp"].between(before_1w, end_inclusive, inclusive="both")]
    final_out.reset_index(drop=True, inplace=True)
    print("final_outを一週間前から今日までにする: \n", final_out)
    
    final_out.to_parquet(PROC_OUT, index=False)
    print(f"[OK] wrote filtered parquet (clean): {PROC_OUT}  rows={len(final_out)}")

if __name__ == "__main__":
    fetch_price()

• いったんcsvを取得してから列名をあてはめるようにしています
• 必要なcsvは過去一週間分なのでフィルターをかけています
• タイムアウト時のデバッグ方法として、以下のファイルを残します
  • screenshot: その瞬間の画面
  • tracing: クリック・DOMの状態遷移のログ（zip）
    ⇒ 今回はcalender_page.pngとtrace.zipを残すようにしています

上記の関数を取得すると以下のデータが取得できます。

• 当年度の電力価格データ（csv）
• 一週間前から今日までの電力価格（parquet）

明日

明日は自動取得したデータをまとめて電力需要の予測を行っていきます！