コーディング面接：データストリームからの中央値の取得

問題と適用されるコンテキスト

2つの操作を持つ MedianFinder を設計します。

• addNum(num) はストリームに整数を追加します。
• findMedian() はこれまでに確認されたすべての値の中央値を返します。要素数が奇数の場合は中央の値を返し、偶数の場合は中央の2つの値の平均を返します。

削除はなく挿入のみと仮定し、findMedian() は少なくとも1回の挿入の後にのみ呼び出されると仮定します。入力には負の数、重複、符号付き32ビット整数が含まれる可能性があり、操作回数は最大50,000回です。目標は挿入あたり O(log n)、クエリあたり O(1)、空間計算量 O(n) です。

例えば、5, 2, 10, 4 を挿入した後の累積中央値は 5, 3.5, 5, 4.5 です。クエリごとにソートするのは正しいものの、クエリあたり O(n log n) のコストがかかります。配列を常に完全にソートされた状態に保つとクエリは O(1) になりますが、中央への挿入によって依然として O(n) 個の要素のシフトが発生します。

2026年現在の一般的な面接対策資料でも、これは2つのヒープを使用する代表的なコーディング問題として扱われ続けています。一般的なソフトウェア、バックエンド、データ、インフラのコーディング面接に適用されます。ここで評価される有用なシグナルは、「最大ヒープと最小ヒープ」というフレーズを覚えていることではありません。クエリからデータ構造を導き出し、両方の不変条件を述べ、固定の要素移動手順によって分割が維持される理由を証明することです。

面接官が評価しているポイント

第1のシグナルは、操作の組み合わせから適切なデータ構造を選択できることです。中央値はソート順の中央にのみ依存するため、全体の順序を完全に保持する必要はありません。O(1) で応答するには、1つまたは2つの中央値の候補が常に直接読み取り可能な位置に露出している必要があります。ヒープの先頭（ルート）は、まさにその境界へのアクセスを提供します。

第2のシグナルは、分割とバランスの両方を維持することです。

1. 最大ヒープ lower が小さい方の半分を格納し、最小ヒープ upper が大きい方の半分を格納し、lower 内のすべての値は upper 内のすべての値以下であること。
2. lower は upper と同じサイズ、またはちょうど1要素だけ多いこと。

どちらか一方の条件だけでは不十分です。サイズが近くても、値が誤った半分に配置されるのを防ぐことはできません。また、正しい分割順序であっても、一方のヒープが極端に大きくなるのを防ぐことはできず、その場合ヒープの先頭は中央を表さなくなります。

第3のシグナルは、正確な計算量分析です。挿入は定数回のヒープ操作を実行し、それぞれ O(log n) です。クエリは1つまたは2つの先頭を読み取るため、O(1) です。この構造は依然としてすべての入力を保持するため、O(n) の空間を使用します。ここでの「ストリーミング」とはオンライン更新を意味し、メモリが定数であることを意味するわけではありません。

最後に、面接官はサンプルの枠を超えた検証を求めます。優れた回答では、最初の要素、偶数と奇数の要素数、重複、すべて負の値、昇順および降順のシーケンス、整数の極値などをテストします。また、ランダムな操作シーケンスを、低速だが確実に正しいソート済みリストモデルと比較検証します。

回答前に確認すべき明確化のための質問

• 挿入のみですか、それとも古い値を削除する必要がありますか？ 挿入のみであれば、通常の2つのヒープで十分です。スライディングウィンドウの場合は遅延削除または順序付きマルチセットが必要です。
• 空のストリームに対してクエリが実行されることはありますか？ この問題文では実行されないとされています。プロダクションAPIでは、空のヒープの先頭を読み取る代わりに、オプショナルな値を返すか明示的なエラーを発生させるべきです。
• 入力は整数ですか、それとも浮動小数点数ですか？ このバージョンでは整数を使用します。浮動小数点数の NaN が許可される場合、値は通常の全順序を形成しないため、拒否するか順序付けのセマンティクスを定義する必要があります。
• 偶数個の場合の中央値はどのように定義されますか？ この問題文では中央の2つの値の算術平均を使用するため、戻り値の型は小数を表現できる必要があります。
• 結果は厳密である必要がありますか？ はい。固定のメモリ予算下での無制限のストリームでは、代わりにおおよその分位数を求める規約が必要になります。
• 平均の計算でオーバーフローは発生しますか？ Pythonの整数はオーバーフローしません。固定幅の言語では、加算と除算の前に両方のオペランドを拡張（キャスト）する必要があります。
• クエリと挿入の比率はどのくらいですか？ 頻繁にクエリが発生する場合は2つのヒープが適しています。すべての入力が到着した後に中央値が1回だけ要求される場合は、収集してソートする方が通常はシンプルです。
• 並行アクセスは必要ですか？ 実装はシングルスレッドです。並行バージョンでは、要素の移動とクエリが両方のヒープの単一の状態を観測できるようにする必要があります。

30秒の回答フレームワーク

「小さい方の半分を lower という最大ヒープで保持し、大きい方の半分を upper という最小ヒープで保持します。lower 内のすべての値は upper 内のすべての値以下でなければならず、lower は同じサイズか、ちょうど1要素多く保持します。挿入時には、まず lower にプッシュし、その最大値を upper に移動して分割順序を回復し、upper の方が大きくなった場合は upper の最小値を戻します。奇数個の場合、中央値は lower の先頭であり、偶数個の場合は両方の先頭の平均です。挿入は定数回の O(log n) のヒープ操作を使用し、クエリは O(1)、空間計算量は O(n) です。」

ステップごとの詳細解説

ステップ1：ベースラインのアプローチを比較し、ボトルネックを特定する。

二分探索を使用すると配列の挿入インデックスを O(log n) で見つけることができますが、O(n) のシフトコストを解消することはできません。通常の平衡木は順序を保持しますが、部分木のサイズ情報がなければ k 番目の要素を直接選択することはできません。2つのヒープは中央値に必要な2つの境界のみを保持するため、この要件に対して最小の完全な構造となります。

ステップ2：中央値を1つまたは2つのヒープの先頭として書き換える。

小さい方の半分を最大ヒープ lower に格納し、その半分の最大値を露出させます。大きい方の半分を最小ヒープ upper に格納し、その半分の最小値を露出させます。lower には1要素多く保持することを許可します。

Odd total:  lower has one extra, median = max(lower)
Even total: heaps have equal sizes, median = (max(lower) + min(upper)) / 2

広く利用可能なPythonの heapq インターフェースは最小ヒープに基づいています。一般的なPythonバージョン間で移植性を保つため、lower には符号を反転した値を格納します。論理的な最大値 x は格納された最小の負の値 -x となるため、-lower[0] は小さい方の半分の最大値になります。

ステップ3：固定のプッシュ、移動、リバランスの手順を使用する。

新しい値の考えられるすべての行き先について分岐する代わりに、常に次の手順を実行します。

1. 符号を反転した num を lower にプッシュする。
2. lower の論理的な最大値をポップし、それを upper にプッシュする。
3. upper の方が大きくなった場合、その最小値を lower に戻す。

import heapq


class MedianFinder:
    def __init__(self) -> None:
        self.lower = []  # Negated max-heap containing the smaller half
        self.upper = []  # Min-heap containing the larger half

    def add_num(self, num: int) -> None:
        heapq.heappush(self.lower, -num)

        largest_lower = -heapq.heappop(self.lower)
        heapq.heappush(self.upper, largest_lower)

        if len(self.upper) > len(self.lower):
            smallest_upper = heapq.heappop(self.upper)
            heapq.heappush(self.lower, -smallest_upper)

    def find_median(self) -> float:
        if not self.lower:
            raise ValueError("median is undefined for an empty stream")

        if len(self.lower) > len(self.upper):
            return float(-self.lower[0])

        return (-self.lower[0] + self.upper[0]) / 2.0

この手順は見かけ上一見余計な移動を行いますが、エラーが発生しやすい複数のケース分けを排除できます。もう1つの有効な実装は、num と -lower[0] を比較し、ヒープを選択してからリバランスする方法です。どちらも漸近的なコストは同じです。面接では、不変条件を確実に証明およびレビューできる方を選択してください。

ステップ4：順序の不変条件を証明する。

挿入前に、lower 内のすべての値が upper 内のすべての値以下であると仮定します。新しい値が一時的に lower にプッシュされた後、誤った半分にある可能性があるのはその新しい値だけです。拡張された lower の最大値をポップします。

• lower に残っているすべての値は、ポップされた値以下です。
• lower のすべての古い値は、すでに upper のすべての古い値以下でした。
• したがって、ポップされた最大値を upper に追加した後も、新しい lower のすべての値は新しい upper のすべての値以下であり続けます。

その移動の後、upper の要素が1つ多くなる場合があります。その最小値を lower に戻しても順序は維持されます。移動された値は upper に残っているすべての値以下であり、かつ古い下限値以上です。その後、ヒープは同じサイズになるか、lower が1つ多くなります。

両方の不変条件は、2つの空のヒープに対して成り立ちます。各挿入でこれらが維持されるため、数学的帰納法により、任意の操作シーケンスの後でもヒープの先頭は中央の位置を表します。

ステップ5：分割をまたぐシーケンスをトレースする。

Insert 5:  lower = [5]       upper = []        median = 5
Insert 2:  lower = [2]       upper = [5]       median = 3.5
Insert 10: lower = [5, 2]    upper = [10]      median = 5
Insert 4:  lower = [4, 2]    upper = [5, 10]   median = 4.5

ヒープの内部配列は完全にソートされているわけではありません。[4, 2] は単に4が最大ヒープの先頭であることを意味します。デバッグのチェックでは、内部配列をソート済みリストとして比較するのではなく、ヒープ順序、2つの先頭の値、およびヒープ間の不変条件を検証する必要があります。

ステップ6：計算量を計算し、より単純なアプローチが優れているケースを特定する。

add_num は最大5回のプッシュまたはポップを実行します。各ヒープ操作は O(log n) であるため、定数回の操作も依然として O(log n) です。find_median は長さとヒープの先頭を O(1) で読み取ります。すべての値はちょうど1つのヒープに存在するため、空間計算量は O(n) になります。

プロダクトが1つのバッチを収集し、最後に1回だけ中央値を要求する場合は、配列を保存してソートする方がコードが短く、連続メモリの挙動も優れている可能性があります。オンラインデータ構造を維持する必要はありません。すべての値が 0 から 100 の固定範囲内にある場合、101個のカウント配列を使用すれば、挿入は O(1)、走査は固定の101個のバケットとなり、その固定ドメインに対してやはり定数時間になります。

ステップ7：確定的なケースとランダムな差分テストで検証を完了する。

最低限、以下をテストします。

ランダムテストでは、生成された各整数を MedianFinder と参照配列の両方に追加します。挿入ごとに参照配列をソートして中央値を計算します。両方の結果を比較し、len(lower) が len(upper) と等しいか1大きいことをアサートします。この低速なモデルは目標のパフォーマンスには適していませんが、正確性のオラクル（検証基準）として非常に優れています。

高品質な回答例

「まず、これが挿入のみの厳密な中央値計算であり、空のストリームに対してクエリが発生しないことを確認します。古いウィンドウ要素を削除する必要がある場合、通常のヒープでは任意の位置の値を効率的に削除できないため、設計が変わります。

定数時間のクエリを実現するために、ソート順の中央が常にデータ構造の境界に露出するようにします。小さい方の半分には最大ヒープ lower を、大きい方の半分には最小ヒープ upper を使用します。重要な不変条件は2つあります。lower 内のすべての値が upper 内のすべての値以下であること、そして lower は同じサイズか1要素だけ多いことです。

挿入時には固定の3ステップの手順を使用します。新しい値を lower にプッシュし、lower の最大値を upper に移動して分割を回復し、upper の方が大きくなった場合は upper の最小値を戻します。これで両方の不変条件が再び成り立ちます。奇数個の場合、lower に余分な値が含まれているため、その先頭が中央値になります。偶数個の場合は、両方の先頭の平均をとります。

挿入は定数回のヒープ操作を実行するため O(log n) です。クエリはヒープの先頭を O(1) で読み取り、すべての値を保持するため O(n) の空間が必要です。単一要素、偶数個、重複、負の値、単調な入力、整数の極値をテストし、各ステップでソートするモデルに対してランダムな差分テストを実行します。値が 0 から 100 に制限されている場合は101個のカウンタを使用し、最後の1回だけクエリがある場合は単純にソートします。」

よくある落とし穴

• ヒープのサイズのみをバランスさせる → 値が分割境界を越えてしまい、先頭が中央の2つの値にならなくなる → 順序とサイズの両方の不変条件を維持する。
• 小さい方の半分を最小ヒープに格納する → その先頭が下位半分の最大値ではなく、全体の最小値になってしまう → 小さい方の半分には最大ヒープを使用する。
• 偶数個のときに一方の先頭のみを返す → 中央値の定義が誤っている → サイズが等しいときは両方の先頭の平均をとる。
• 型変換の前に固定幅整数を加算する → 2つの大きな値が先にオーバーフローする可能性がある → 加算および除算の前に両方のオペランドを拡張（キャスト）する。
• ソート済み配列への挿入を O(log n) と主張する → インデックスの検索は高速だが、シフト操作に依然として O(n) かかる → 探索コストと変更コストを区別する。
• Pythonのヒープ配列を完全にソートされているものとして扱う → デバッグ用のアサーションが無効になる → ルートおよび親子のヒープ特性のみに依存する。
• 空のヒープからインデックス0を読み取る → 不明確な境界で失敗する → 空のクエリを禁止するか、明示的にオプショナルな値を返す。
• オンラインアルゴリズムが定数空間を使用すると主張する → 両方のヒープがすべての入力を保持する → 厳密な中央値には O(n) の空間が必要であることを明記する。
• スライディングウィンドウに同じコードを再利用する → 期限切れの値が先頭に残り、結果を破損する可能性がある → 遅延削除と有効サイズの追跡を追加するか、順序付きマルチセットを使用する。
• サンプルケースのみをテストする → 符号反転、重複、リバランスのバグが顕在化しない可能性がある → エッジケースとランダムな差分テストを組み合わせる。

フォローアップの質問と回答

フォローアップ1：すべての整数が0から100の間である場合、何が変わりますか？

101個のカウント配列と総要素数を保持します。挿入は O(1) で1つのバケットをインクリメントします。クエリを実行するには、中央の1つまたは2つの順位に達するまでバケットを走査します。走査と空間はこの固定ドメインに対して定数です。範囲が入力とともに拡大する場合、走査は範囲サイズ R に対して O(R) となり、もはや定数とは言えなくなります。

フォローアップ2：値の99%が0から100の間で、残りが任意の値である場合はどうなりますか？

範囲内の値には101個のカウンタを保持し、0未満および100を超える値には順序統計量データ構造を保持します。それらのカウントによって、目的の順位が下位の外れ値、固定範囲、上位の外れ値のどこにあるかを判断し、関連する構造内で選択を行います。通常のヒープは任意の順位選択をサポートしていないため、「99%」という記述だけでは定数時間のクエリを正当化できません。敵対的な入力プレフィックスによって、中央値の順位が外れ値の中に位置する可能性があります。

フォローアップ3：直近 k 個の値の中央値を計算するにはどうすればよいですか？

ウィンドウの移動には、退出する値の削除が必要です。二分ヒープは任意のエントリを効率的に見つけることができません。一般的な解決策は、遅延削除カウントマップを追加し、両方のヒープの有効サイズを追跡することです。退出する値に論理的な削除マークを付け、それが先頭に達したときにのみ物理的にポップします。中央値を読み取る前に両方の先頭をプルーニング（不要データの除去）します。更新はならし O(log k) で、先頭の参照は O(1) のままです。言語がサポートしている場合は、重複を許容する平衡マルチセットを使用する方がシンプルです。

フォローアップ4：固定メモリのアルゴリズムで、無制限のストリームの厳密な中央値を返すことはできますか？

一般に、任意の整数ストリームに対しては不可能です。破棄された過去の値が後で中央の順位を決定する可能性があるためです。要件を近似分位数に変更し、明示的な順位誤差の保証、信頼度要件、マージ動作を持つ分位数スケッチ（Quantile Sketch）を使用する必要があります。これは厳密な2ヒープ構造とは異なる回答になります。

フォローアップ5：並行な挿入とクエリをサポートするにはどうすればよいですか？

2つのヒープは1つの論理的状態を形成します。最も単純で正しい拡張は、完全な add_num および find_median 操作を同じミューテックスで保護し、値が lower を離れてから upper に入るまでの瞬間をクエリが観測できないようにすることです。読み取りが多いサービスでは、不変の中央値スナップショットを公開することもできますが、スナップショットの間隔によって鮮度のトレードオフが生じ、これはAPIの規約に含める必要があります。

フォローアップ6：複数のシャードからの中央値を結合してグローバルな中央値を算出できますか？

いいえ。シャードの中央値は、そのシャードのサイズと分布情報を失っています。シャードの中央値の加重平均であってもグローバルな中央値にはなりません。厳密な結果を得るには、有界ドメイン上のカウントを集約して分散選択を実行するなど、グローバルな順位に応答できる構造が必要です。近似的な結果であれば、マージ可能な分位数サマリーを使用できます。グローバルな構造を決定する前に、精度とレイテンシの要件を選択する必要があります。