取引におけるPython – データ分析、自動化、研究

Pythonは市場データの整理、取引日誌の分析、数式を手でコピーしない計算の反復に役立ちます。表計算では規模が足りなくなったり、管理が難しくなったりしたときに特に有用です。本記事では環境設定から、データを読み込み、品質を確認し、結果を保存するスクリプトまでを説明します。また、ノートブックの研究と取引口座に接続するソフトウェアの違い、接続に必要な安全措置も扱います。

1. Pythonが役立つ場面と、解決策に見えるだけの場面

表計算で取引日誌を扱えるのは、利用者の忍耐が続く間です。その境界はPC、数式数、シート構成に依存し、切りのよい取引数では表せません。しかし、ファイル形式の仕様に書かれた、他に依存しない厳密な上限はあります。Excelの一枚のワークシートは1,048,576行、16,384列です ^[16]。一通貨ペアの一年分のティック履歴はその何倍にもなり、工夫では回避できません。

ここでPythonは魔法ではありません。表計算では解決できない、あるいは何時間もかかる四つの具体的問題を解決します。

• データ規模 – 上限はPCのメモリと保存形式であり、シートの行数ではありません
• 反復可能性 – 同じスクリプトを明日同じデータで動かせば同じ結果になります。手で数式を引き延ばす表計算には、その保証がありません
• 手作業なしの時間間隔変更 – OHLCVを保ってM1をH4に集約するには、一回の resample 呼び出しで済み、補助数式の列は不要です
• データ取得と整理の自動化 – 定時実行スクリプトが、不在でも取得、確認、保存を行います

それだけです。Pythonは市場の優位性を与えず、戦略品質を上げず、ブローカーの執行条件も変えません。問いと答えの距離を縮めるので、アイデアが通用しないことも早く分かります。

Pythonがしないこと。 超低遅延取引には不向きです。マイクロ秒が重要な領域では、ガベージコレクターを備えたインタープリター言語は適しません。しかし「Pythonは取引には遅すぎる」も誤りです。次のH1足の始まりで入る戦略なら、計算時間は、注文がネットワークとブローカーサーバーを通る時間に比べて無視できます。

ベクトル化の高速化は魔法ではなく仕組みによるものです。列全体の演算はインタープリターのループでなく、コンパイル済みnumpyコードで走ります。差の桁はデータ、演算、機器に依存します。具体値を示すなら、自分のPCで標準ライブラリの timeit モジュールを使って測定し、 ^[4] ライブラリ版とともに記録してください。環境のない他人のベンチマークの引用は測定ではありません。

トレーダーの分析ツールの比較：

表の最後の行の違いは、見た目以上に重要です。Pine ScriptはTradingView内、MQLはMetaTrader端末内、Pythonは利用者が起動した場所で動きます。その自由は、ライブラリ版、OSのタイムゾーン、稼働維持、そもそも口座に触れる権限があるかという環境の責任を伴います。

2. 再現可能な環境：バージョン、venv、requirements

他人のスクリプトが「動かない」よくある理由はコードでなく、作成者が別のライブラリ版で実行し、記録しなかったことです。市場データ分析では単なる不便ではなく、再現できないため結果が無効になります。

再現可能な環境には六つを示せます。Python版、隔離方法、固定した依存版、期間付きのデータ出所、タイムゾーン、そしてランダム性が結果に影響するすべての箇所の乱数シードです。欠ければ測定ではなく物語になります。

仮想環境は任意ではありません。 ライブラリを全体に入れると、あるプロジェクトの更新が別のものを壊します。隔離は無料で十数秒、標準ライブラリの venv モジュールを使います ^[4]:

# Windows、PowerShell
py -3.12 -m venv .venv
.venv\Scripts\Activate.ps1

# macOS / Linux
python3.12 -m venv .venv
source .venv/bin/activate

# 両OS共通
python -m pip install --upgrade pip
python -m pip install pandas numpy matplotlib requests python-dotenv
python -m pip freeze > requirements.txt

最も省かれがちなのは最後の行です。 pip freeze は、知らなかった依存も含め、実際に入った全パッケージの正確な版を記録します。そのファイルをコードと並べてリポジトリに置きます。別PCでの再現は、推測でなく、 python -m pip install -r requirements.txtで行えます。

今回のオフライン検証に使った環境 は第4・9節の計算例に対するもので、Windows 11、Python 3.12.10、pandas 3.0.5、NumPy 2.5.3です。確認日は2026年10月8日です。以下は計算に関わるパッケージだけの限定的な一覧で、PCの全パッケージの書き出しではありません^[1][3]。

# ローカル計算検証に使った依存のバージョン
numpy==2.5.3
pandas==3.0.5
python-dateutil==2.9.0.post0
six==1.17.0
tzdata==2026.4

# インタープリター
Python 3.12.10

pandasは他のパッケージも使うため、環境説明は一ライブラリの版だけでは終わりません。依存によりPC間で差が生じます。上のインストールコマンドは準備手順の例で、今回の監査で実施したインストール記録ではありません。自分の環境で完全な requirements.txt を作り、方法には python -m pip freeze を使って、コードと保管してください。

このファイルは記事全体の完全な環境説明でもなく、計算部分の環境だけを表します。第7節のHTTPエラー処理は、 requests 2.34.2を追加した環境で実行しました。第8節の安全ゲート試験は標準ライブラリだけで動きます。 matplotlib と python-dotenv の版は、実行済み例で使っていないため示しません。「体裁のため」に載せると誰も確認していない数値を書くことになります。全体の環境を作るなら、上のインストールコマンドを出発点に、自分の環境で pip freezeを固定してください。自分の結果について意味を持つ依存ファイルはそれだけです。

ノートブックか.pyファイルか？ 両方ですが用途は違います。何を探すか未定なら、セルを動かし結果を見てパラメータを変えるJupyterは便利です。ただし、変数の状態がファイル内の記載順でなく、セルの実行順に依存するため、執行コードの置き場には不適格です。セルを1–3–2の順で実行しても、順番通りに実行したものと見た目は同じです。口座に触れるコードは通常の .py ファイルにし、上から下へ実行し、テストとログを備えます。

AnacondaとGoogle Colab は固有の代償を伴う近道です。Anacondaは多くのパッケージを一度に入れ、独自の依存管理を持つため開始しやすい一方、 conda と pipを混ぜると正確な環境説明が難しくなります。Colabはインストール不要でブラウザーで動き、学習に適しますが、毎回環境が再作成され、利用者の関与なくライブラリ版が変わります。一年後にも再現する資料には、どちらも追加の規律が必要です。

最初のライブラリ – 四十ではなく四項目（最初は一組なのでパッケージは五つ）：

• pandas と numpy – データ構造と計算
• matplotlib – グラフ
• requests – APIとのHTTP通信
• python-dotenv – リポジトリ外のファイルから設定を読み込む

テクニカル指標は必要になってから追加します。古い案内への修正です。 TA-Lib はCライブラリ込みの公式バイナリwheelをPyPIで公開しています。対応するOS、アーキテクチャ、Pythonの組み合わせなら、単に pip install TA-Lib として入れられ、非公式ダウンロードも自前のコンパイルも不要です^[9]。ただし全PCで動く約束ではないので、該当wheelを確認してください。2026年10月8日確認のメタデータでは、TA-Lib 0.8.1と pandas-ta は2025年9月公開の0.4.71b0のままです^[13]。後者は、 numba==0.61.2を固定するため、pandasの条件だけの確認では不十分です。2026年8月22日のPython 3.14.3での過去の導入試験は、その依存で失敗し、 only versions >=3.10,<3.14 are supportedと表示されました。この観測は、現在のPython 3.12.10の例に自動的には当てはまりません。今回の監査はPyPIメタデータを読み、TA-Libやpandas-taを導入・実行してはいません。したがって、隣の確認原則だけでは狭すぎます。最近の公開日と互換性のあるpandas条件だけで、全依存の導入・実行は証明できません。固定されたコンパイル済みライブラリ、使用インタープリター用wheel、小規模データ試験も確認します。最新二版のメタデータは全履歴を示さないので、それ以前の開発の空白はリポジトリで評価してください。

乱数シード。 標本分割、ブートストラップ、シミュレーション、モデル初期化など、結果が乱数に依存するすべての箇所でシードを設定・記録します。numpyなら、 rng = np.random.default_rng(20260822) ^[3]とし、そのオブジェクトだけを使います。なければ同じスクリプトを二度実行して結果が変わり、どちらを説明しているか不明になります。

3. pandasとnumpy：OHLCVデータ、時刻インデックス、タイムゾーン

numpyは数値配列の計算、pandasは構造を扱います。DataFrameは名前付き列と、多くは時刻のインデックスを持つ配列です。市場データでは時刻インデックスが計算自体より重要です。意図した対象を計算しているかを決めるからです。

読み込み – pd.read_csv() は数百万行のファイルを扱い、区切り文字を認識し、列を指定すれば日付も解析します。ただし標準ではタイムゾーンを知らないので、最初に確定する必要があります。

時間間隔の変更 – OHLCVを保ってM1をH4に集約するのは一回の呼び出しです。 df.resample('4h').agg({'open':'first', 'high':'max', 'low':'min', 'close':'last', 'tick_volume':'sum'})。設定されないことが多く、解釈を決める二引数に注意します。 label は窓の開始と終了のどちらにラベルを置くかを決め、 closed は区間のどちらの端を含むかを決めます。多くの頻度では標準で左端にラベルが付き、 ^[1]足の時刻は開始時刻を表します。pandas 3.0.5で、 resample('4h') を午前零時からの八本の一時間足に適用すると、ラベルは00:00と04:00です。シグナルは終値基準なのにデータが始値時刻なら、気づかず結果全体を一区間ずらします。

移動統計 – df['close'].rolling(20).mean() は、 移動平均を求め、, .rolling(20).std() は、 ボリンジャーバンド用の標準偏差を求め、, .ewm(span=20).mean() は指数移動平均を求めます ^[1]。いずれも現在行から見て過去のデータだけで計算され、その性質が他のすべての正しさの基礎になります。

抽出 – 複合条件は一度書くだけです。 df[(df['sma20'] > df['sma50']) & (df['rsi'] < 30)]。各条件の括弧は必須です。 & はPythonの比較演算より優先されるため、省くと黙って誤答するのでなくエラーになります。この点ではよいことです。

タイムゾーンは細部ではなく判断。 ブローカーデータは通常サーバー時刻、外部提供元は多くがUTC、OSは選択した地域の夏時間変更を含むローカル時刻の場合があります。この三時計を混ぜると、「ロンドン時間」の分析が別の時間を分析してしまいます。

まず出所のゾーンを確定し、計算はUTCで行います。明示的なオフセットか、 Z付きの時刻には、 pd.to_datetime(column, utc=True)を使います。オフセットのないサーバー時刻は先に解析し、確認したゾーンを、 tz_localize() で割り当て、曖昧な時刻と存在しない時刻を処理してから、 tz_convert("UTC")を適用します。 utc=True だけではゾーンなし値をUTCとみなすだけで、ブローカーのゾーンは推定しません。 Europe/London への変換は表示時だけにします^[1]。明示的なゾーン付きインデックスと出所の記録があれば、系列を一貫して結合できます。

欠落データ。 OHLCV履歴には週末、休日、技術的中断、非配信による穴があります。pandasの、 df.ffill()は直前値で補いますが、価格では通常よくありません。変動ゼロの足を作り、標準偏差を下げるからです。欠落期間を取り除き、その数を数える方がよいでしょう。

複数銘柄を同時に扱う。 通貨ポートフォリオの分析には、pandasの階層インデックス（MultiIndex）が使えます。一つのDataFrameに複数ペアを入れ、ペアと時刻で索引します。全系列を同じゾーン、同じ時刻格子にそろえれば、共通の時間軸で相関を計算できます。そろえなければ、互いにずれたデータの相関を計算することになります。

結果の保存。 df.to_csv() は人が読める汎用形式です。 df.to_parquet() はファイルを小さくし、読み込みを速め、列型とインデックスのゾーンも保ちます ^[1] 。CSVはすべて文字列として保存するため、再読込時に再解析します。大量の作業データにはParquetが便利ですが、追加の書込エンジンが必要なので、 requirements.txt.

4. 最小例：CSV → UTC → リターン → 欠落確認 → 保存

最初のスクリプトは注文せず、ブローカーに接続する必要もありません。役立ち、検証でき、完全に安全なことをします。ファイルを読み、時刻を整え、リターンを計算し、データの問題を明示して結果を保存します。

前提。 CSVの列は、 time,open,high,low,close,tick_volumeです。 time 列はゾーン付きISO 8601形式で、Z はUTCです。時間間隔は毎時と宣言します。実際のエクスポートで起こる三つの欠陥、重複時刻、欠落した一時間、空の出来高セルを意図的に含めます。六行をコードに直接入れるので、ファイルを取得せず全く同じ結果を再現できます。再現できない結果は結果ではありません。環境はPython 3.12.10、pandas 3.0.5、numpy 2.5.3、Windows 11です。 ［時点：2026-10-08］.

"""最小例：CSV -> UTC -> 対数リターン -> 欠落確認 -> 保存。"""
import io
from pathlib import Path

import numpy as np
import pandas as pd

# ファイルを取得せず再現できるよう、データをコード内に置く。
# 六行に三つの意図的な欠陥：01:00の重複、
# 03:00の欠落と02:00の出来高セル空白。
CSV_DATA = """time,open,high,low,close,tick_volume
2026-06-01T00:00:00Z,1.08140,1.08170,1.08131,1.08153,812
2026-06-01T01:00:00Z,1.08153,1.08190,1.08150,1.08176,640
2026-06-01T01:00:00Z,1.08153,1.08195,1.08149,1.08182,655
2026-06-01T02:00:00Z,1.08182,1.08188,1.08130,1.08141,
2026-06-01T04:00:00Z,1.08120,1.08125,1.08040,1.08050,903
2026-06-01T05:00:00Z,1.08050,1.08061,1.07980,1.07993,774
"""

OUTPUT = Path("EURUSD_H1_returns.csv")
INTERVAL = "1h"  # 宣言した入力データの頻度


def load_data(source) -> pd.DataFrame:
    """CSVを読み、UTCのゾーン付き索引を設定する。

    意図的に並べ替えない。行の順序自体がエクスポート品質の情報になる。
    並べ替えは報告後に行う。
    """
    df = pd.read_csv(source)
    # utc=Trueでゾーンを統一する。なければゾーンなし索引になる場合がある
    df["time"] = pd.to_datetime(df["time"], utc=True, format="ISO8601")
    return df.set_index("time")


def quality_report(df: pd.DataFrame, interval: str) -> dict:
    """計算前に確認すべき件数。"""
    expected = pd.date_range(df.index.min(), df.index.max(), freq=interval, tz="UTC")
    return {
        "rows": int(len(df)),
        "index_duplicates": int(df.index.duplicated().sum()),
        "missing_cells": int(df.isna().sum().sum()),
        "missing_timestamps": int(len(expected.difference(df.index))),
        # 並べ替え前に測定。sort_index()後では常に
        # trueとなり、時系列の乱れを検出できない。
        "file_chronology_ok": bool(df.index.is_monotonic_increasing),
    }


def add_returns(df: pd.DataFrame) -> pd.DataFrame:
    """対数リターン = ln(close_t / close_(t-1))。単位は自然対数で無次元。"""
    df = df[~df.index.duplicated(keep="last")].copy()
    df["log_return"] = np.log(df["close"]).diff()
    return df


def main() -> None:
    df = load_data(io.StringIO(CSV_DATA))  # 実務では：load_data(Path("EURUSD_H1.csv"))
    print("品質確認：", quality_report(df, INTERVAL))
    df = add_returns(df.sort_index())
    print("重複除去後の足数：", len(df))
    print("対数リターン合計：", round(float(df["log_return"].sum()), 6))
    print("対数リターン標準偏差（1h・2h間隔）：",
          round(float(df["log_return"].std()), 6))
    df.to_csv(OUTPUT)
    print("保存先：", OUTPUT)


if __name__ == "__main__":
    main()

実行結果 は、コード内のデータと説明した環境によるもので、そのまま再現できます。

品質確認： {'rows': 6, 'index_duplicates': 1, 'missing_cells': 1, 'missing_timestamps': 1, 'file_chronology_ok': True}
重複除去後の足数： 5
対数リターン合計： -0.00148
対数リターン標準偏差（1h・2h間隔）： 0.000467
保存先： EURUSD_H1_returns.csv

解釈。 スクリプトは計算前に、仕込んだ三つの欠陥を検出しました。それが価値のすべてです。対数リターンの合計は−0.00148。重複除去後の五本の足、すなわち連続した観測価格の間の四区間では、対象期間の下落は約0.15パーセントです。六はファイルの行数であり足数ではありません。一行は重複です。標準偏差0.000467の単位は、 H1足一本当たりで、年次でも日次でもありません。間隔を示さない比較は無意味です。対数リターンは無次元で時間に対して加算可能なので、単純リターンと違い、掛けずに足します。

限界。 五本は仕組みの例であり統計標本ではなく、市場について結論は出ません。バックテストではないため、スプレッド、手数料、 スリッページをモデル化せず、口座損益ではなく価格リターンを計算します。欠落時刻検出は、本当に宣言した毎時データであることが前提です。週末を含めば「欠落」が多くても正常なので、警報の閾値は取引カレンダーで決めます。尺度自体にも罠があります。スクリプトは穴を検出しても埋めないため、その前後のリターンは一つでなく二つの間隔をまたぎます。この例では四リターンの一つが該当するのに、全体を「H1足当たり」と呼んでいます。これは正確な尺度ではなく簡略化です。正しく扱うには、完全な毎時格子に再索引して穴を欠損のまま残すか、異なる時間幅の混在を測定の限界として明記します。

並べ替え前に時系列を確認する理由。 もし、 load_data() が最後に、 sort_index() を呼ぶなら、 file_chronology_ok は常にtrueとなります。報告書に掲げた検査が何も検出できず、順序の乱れたファイルも痕跡なく通ります。失敗し得ない検査は偽の安心を与えるため、検査なしより悪いものです。そのためファイルから読み込んだ順序を測り、報告後に並べ替えます。

検証状況： この部分は記載の環境で実行し、上は実際の出力です。別のpandas版では辞書の表示や丸めが異なる場合があります。

5. 可視化：一つのグラフに一つの問い

成績を要約した一数値は、いつ損をし、どれだけ続き、全成績が三取引に依存するかを隠します。グラフは報告書の飾りでなく、表では答えられない問いをデータに投げる方法です。

matplotlib はPythonの基本グラフライブラリで、静的で文法はやや扱いにくいものです。 plotly は対話型です。点に重ねて日時と値を見たり、拡大したり、HTMLに出力できます。非常に大きいデータでは遅くなる場合があり、別途インストールも必要です。ノートブックの探索にはplotly、報告書へ保存する図にはmatplotlibが便利です。

異なる問いに答える、描けると役立つ四つの図：

1. 資産曲線 – 時間に沿った累積損益。「増えたか、いつ増えたか」に答え、「いくら儲かるか」には答えません
2. ドローダウン図 – 各時点の過去最高値からの距離。最大下落と回復時間を示します。残高か時価資産か、％か通貨額かを必ず示します
3. セッションのヒートマップ – 時間帯と曜日別の成績。先にゾーンを確定しないと、存在しない時間を説明します
4. リターン分布 – 個々の取引損益のヒストグラム。単一のリスク指標では見えない厚い裾と非対称性を示します

DataFrameからローソク足を描くなら、 mplfinanceというmatplotlib上の層があります。ただし第2節の原則通り注意が必要です。最終公開は2023年8月で、まだプレリリース番号（0.12.10b0）です ^[13] ［時点：2026-10-08］。それだけで失格にはなりませんが、 requirements.txtに加える前に、自分のデータ、matplotlib版、pandas版で動くか確認します。新しい公開版が代わりに確認してくれるわけではありません。スライダーや選択欄のある簡単な画面にするなら、 streamlitという定期的に公開されるプロジェクトがあり、 ^[13]起動方法は、 streamlit run dashboard.pyです。どちらも時間を節約する一方、 requirements.txtの依存を増やすため、意識して追加します。

実用的な原則： どんな問いに答えるグラフかを一文で言えなければ不要です。問いのない図は装飾であり、成績分析の装飾は、データにない結論を示唆して高くつく場合があります。

6. Pythonのバックテスト：コードと方法論の役割

ここには見落としやすい境界があります。Pythonの責任はテストの、 実装 です。データ読込、シグナル計算、正しい期間への損益割当、コスト控除、指標計算を担います。責任を持たないのは、 方法論です。標本分割、試した設定数、執行モデル、頑健性評価を指します。これは別記事の、 デモ口座、バックテスト、フォワードテスト の主題であり、優れたライブラリでも代替できません。

pandasだけではバックテストになりません。 シグナル列とリターン列を掛ければ損益列はできますが、判断と執行の間の事象を何もモデル化しません。エントリー時点、価格、スプレッド、手数料、スワップ、部分約定、拒否です。アイデア選別段階では許容できても、成績ではなく簡略化と呼ぶ必要があります。

自作ループの先には、Python環境に三種類の既成ツールがあります。表では性能の約束でなく動作方式を比べます。所要時間はデータ、機器、版に依存するので、必要なら自分で測り環境を記録します。

最後の行は形式ではありません。各基盤は約定時点と価格を仮定し、戦略ロジックでなく、その仮定が成績を決めることがよくあります。信頼する前に、エントリー価格、コストの徴収方法、足間のギャップの扱いを資料で確認します。

コードの自由は同時にリスク。 内蔵テスターは一定のルールを強制し、一部の誤りを防ぎます。自作コードは何も守りません。訓練・検証データの分離を尋ねず、まだ存在しなかった価格の利用に気づかず、設定の試行数も数えません。責任はすべて作成者に移ります。そのため次節以降は、コード内で気づかず壊せるものを扱います。

7. ブローカーAPI：認証、制限、エラー、テスト環境

「APIがある」を「スクリプトから口座に無制限でアクセスできる」と読むことがありますが、違います。APIには認証、権限、要求頻度制限、定義されたエラーコードがあります。テストと本番に分かれる場合もありますが常にではなく、分け方は資料で調べる別の問題です。最初の一行を書く前に九つ答えられるべきです。認証方式、トークン権限、一秒の要求数、テスト・本番の区別、返すデータ型、エラーコード、地域制限、接続維持方針、そもそも発注権限があるかです。

制限は具体的に公開されるが、何の制限かを読む必要がある。 多くの案内ではこの区別が消えますが、ループ設計を左右します。OANDA v20の二つの数値は別の二文書にあり、意味も異なります。開発者案内は、 強制される 値としてREST API「一秒120要求。超過にはHTTP 429」を示し、ほかに二十のアクティブ配信、毎秒二つの新規接続を定めます。いずれも送信元IP単位です ^[15]。対してベストプラクティス文書は、 推奨として、確立済み接続で毎秒百要求、新規接続は毎秒二つまでに抑えるよう述べます ^[5]。 ［時点：2026-08-22］

毎秒百は推奨する上限で、百二十を超えるとサーバーが429を返します。「100/sが最大」と考えて保守的に動かす害はありませんが、これを強制上限と取り違え、429を処理しなければ最初の超過で落ちます。推奨と強制の境界は違うからです。一社の数値であり市場の規則ではなく、他社では異なり、 ［機能はブローカー／口座による］という条件があります。ポーリングのループを書く前に自分の提供元の資料を確認します。

テスト環境は別モードではなく別ホスト。 OANDAではアドレス自体に区別があり、 api-fxpractice.oanda.com が練習用、 api-fxtrade.oanda.com が本番用で、ストリームにも対応する二つのアドレスがあります ^[15] ［時点：2026-08-22］。本番ホストに追加条件を設け、コードで誤りを検出できる利点があります。一アドレスでトークンだけを区別する提供元では、設定だけが防壁になります。

Pythonを口座につなぐ代表的な四経路。違うのは高度さでなく構造です。

為替以外では、 ccxt も知っておくと有用です。暗号資産取引所への接続を統一するライブラリで、個人向けブローカーのFX/CFD口座には適用しませんが、万能と誤解されることがあります。

cTrader。 cTraderのアルゴリズムはcTrader Algoで書き、C#とPythonの両方を標準サポートします。cBot、指標、プラグインも含みます ^[8] ［時点：2026-08-22］。外部プロセスがAPIに接続するのではなく、基盤内でコードが動く別モデルです。開発が続いているため、機能範囲と起動方法を提供元の現行資料で確認します。

鍵とパスワード：コードに入れない

スクリプトに書いたAPI鍵は、リポジトリ、バックアップ、版履歴にも移り、行を消してもGit履歴には残ります。公開リポジトリはそのような文字列を自動走査されます。秘密の置き場は環境変数、リポジトリ外に置いて、 .gitignoreに登録した設定ファイル、またはそれらを通じて値を供給する秘密管理サービスだけです。

"""設定と秘密：機密情報をコードファイルに入れない。"""
import os
from dataclasses import dataclass

HOSTS = {
    "practice": "https://api-fxpractice.oanda.com",
    "live": "https://api-fxtrade.oanda.com",
}


class MissingConfiguration(RuntimeError):
    """開始に必須のデータが欠けた場合に発生する。"""


@dataclass(frozen=True)
class Configuration:
    token: str
    account: str
    environment: str

    @property
    def host(self) -> str:
        return HOSTS[self.environment]


def _required(name: str) -> str:
    value = os.environ.get(name, "").strip()
    if not value:
        # 既定値を置かないのは意図的。空トークンでは停止せず、
        # 認可なしで要求を送ってしまう。
        raise MissingConfiguration(f"環境変数がありません：{name}")
    return value


def load_config() -> Configuration:
    environment = os.environ.get("BROKER_ENV", "practice").strip().lower()
    if environment not in HOSTS:
        raise MissingConfiguration(f"BROKER_ENVは次のいずれかにしてください：{sorted(HOSTS)}")
    return Configuration(
        token=_required("BROKER_API_TOKEN"),
        account=_required("BROKER_ACCOUNT_ID"),
        environment=environment,
    )

三つの判断は意図的です。標準環境は本番でなく練習なので、設定ミスは実口座でなくデモに向かいます。必須環境変数がなければ空文字を使わず停止します。トークンの初期値はコメントにも置きません。コメント内の秘密も漏洩した秘密だからです。

先述の、 python-dotenv はこのコードを置き換えず、その前に働きます。 load_dotenv() を、 from dotenv import load_dotenvの後に呼ぶと、リポジトリ外の、 .env をプロセス環境変数に読み込み、その後で、 load_config() が読みます。これで設定関数は、出所がファイル、OS設定、秘密管理のどれかを知る必要がありません。 .env 自体は設定ディレクトリ全体とともに、 .gitignore に入れます。

データ取得：タイムアウト、再試行、応答コード

時間制限のない要求は無期限に待ち、全体を停止させ得ます。無制限の再試行は頻度枠を消費し、遮断を招きます。以下は読取の最低限の防護と、後に口座を守る区別を示します。再試行してよいのは、繰り返してもブローカー側の状態を変えない冪等操作だけです。

import logging
import time

import requests

TIMEOUT_S = 10.0        # 接続・読取の無通信秒数。処理全体の期限ではない
MAX_ATTEMPTS = 4            # 初回を含む試行回数
BACKOFF_S = 1.0         # 指数的な待機の基準秒数

log = logging.getLogger("data")


class ApiError(RuntimeError):
    """再試行では直せないエラー。"""


def fetch_candles(session: requests.Session, host: str, token: str,
                   instrument: str, parameters: dict) -> dict:
    """ブローカー側を変えないGETだけを再試行する。"""
    url = f"{host}/v3/instruments/{instrument}/candles"
    headers = {"Authorization": f"Bearer {token}"}

    for attempt in range(1, MAX_ATTEMPTS + 1):
        delay = BACKOFF_S * 2 ** (attempt - 1)
        try:
            response = session.get(url, headers=headers, params=parameters, timeout=TIMEOUT_S)
        except requests.RequestException as error:
            if attempt == MAX_ATTEMPTS:
                raise ApiError(f"接続できません（{MAX_ATTEMPTS}回の試行後）") from error
            log.warning("試行%s: %s", attempt, error)
            time.sleep(delay)
            continue

        if response.status_code == 429:
            # 頻度制限超過。ブローカーが返すRetry-Afterは、
            # 自分の推測より信頼できる。
            try:
                wait = float(response.headers.get("Retry-After", delay))
            except (TypeError, ValueError):
                # 仕様では秒数だけでなくHTTP日付も許される。
                # この分岐がないとfloat()で全取得が停止する。
                wait = delay
            log.warning("頻度制限。%.1f s待ちます", wait)
            time.sleep(wait)
            continue

        if 500 <= response.status_code < 600:
            log.warning("サーバーエラー%s。再試行します", response.status_code)
            time.sleep(delay)
            continue

        if response.status_code != 200:
            raise ApiError(f"HTTP {response.status_code}: {response.text[:200]}")

        try:
            data = response.json()
        except ValueError as error:
            raise ApiError("不正なJSON応答") from error
        if not isinstance(data, dict):
            raise ApiError("JSON応答はオブジェクトである必要があります")
        if not isinstance(data.get("candles"), list) or not data["candles"]:
            raise ApiError("足がありません。銘柄と期間を確認してください")
        return data

    raise ApiError(f"有効な応答がありません（{MAX_ATTEMPTS}回の試行後）")

検証状況：ブローカーに接続せずオフラインで試験済み。 公開した、 fetch_candles() 関数は、実サーバーでなくメモリ内のHTTPセッションモックで実行しました。九つのシナリオは、429と、 Retry-After の秒指定から503を経て成功する一つのケース、空の足一覧、404、HTTP日付、試行上限、接続例外、不正JSON、非オブジェクト応答、不正な、 candles 型を含みます。 sleep() 呼び出しは要求待ち時間の記録に置換しました。ソケットも開かず、ネットワーク時間も測りません。環境はPython 3.12.10、requests 2.34.2、Windows 11、 確認日2026-10-08です。実際のシナリオ報告は次の通りです。

A: 429秒指定 -> 503 -> 200: calls=3; waits=[3.0, 2.0]; 足一本
B: 足一覧が空: calls=1; waits=[]; 想定エラー
C: HTTP 404: calls=1; waits=[]; 想定エラー
D: HTTP日付の代替処理: calls=2; waits=[1.0]; 足一本
E: 試行上限: calls=4; waits=[0.0, 0.0, 0.0, 0.0]; 想定エラー
F: 通信障害: calls=4; waits=[1.0, 2.0, 4.0]; 想定エラー
G: 不正JSON: calls=1; waits=[]; 想定エラー
H: オブジェクト以外のJSON: calls=1; waits=[]; 想定エラー
I: candlesの型が不正: calls=1; waits=[]; 想定エラー

この結果で確認できること、できないこと。 用意した九場面の期待動作を確認します。Aは429でコードの一秒でなくサーバー指定三秒を選ぶこと、Dは、 float() がHTTP日付を読めない場合の代替処理を確認します。Fは、 timeout=10 の受け渡し、四回の試行、模擬通信障害を例外で終えることを確認します。実接続の期限は立証しません。requestsのtimeoutは処理全体の期限ではなく、ここでは待機もモックだからです。B、C、G–Iは空や不正応答を拒否します。APIの利用可能性、全フィールド、特定社のネットワーク挙動は検証しません。

ただしブローカー自体については何も確認しません。エンドポイント、パラメータ名、応答構造、そもそも、 Retry-After を返すかは特定APIの性質です ［機能はブローカー／口座による］。資料確認と練習口座での試験が必要です。モックが証明するのは自分のコードの正しさで、他者のサーバーに対する前提の正しさではありません。その区別が試験の価値です。

遅延は引用でなく測定する。 案内の「このAPIは50–200 ms、あちらは100–500 ms」は、場所、回線、時間帯、測定法がなく無価値です。他人の数値を写さず、データ応答時間、送信から受付確認、約定確認までの時間、判断時のデータ年齢を分けます。四つの異なる量で、一般に「遅延」と呼ぶのはその一つだけです。自分の経路で各々を測り、ログに記録し、日付、ブローカー、試行数とともに示します。ICMP pingはそのどれでもありません。

ポーリングかストリーミングか。 定期REST照会は間隔で古さが決まり、五秒ごとなら最悪五秒前の価格で判断します。頻繁にすると頻度枠を消費します。APIに応じたWebSocketやHTTP配信は、更新時にサーバーが送る逆の仕組みです。足内の反応が重要、または照会が枠を使い切る場合に意味があります。次のH1始値で入るなら、単純で保守しやすいポーリングで十分です。一方、配信は三つの問題を追加します。第一は、 再接続後の欠落です。三十秒切れてもソケットからその三十秒は再送されず、別の履歴要求で補って初めて完全な系列になります。第二は、 死んだ接続の検出です。送信停止したソケットは、静かな市場とコード上では同じです。ハートビートやkeep-aliveと、自分の無通信カウンターで切断を判定し、再接続します。第三は、 順序と重複です。再開時に処理済みのメッセージが再度来ることがあります。注文と同じく、イベントIDと登録済み確認で対処します。

データと注文の制限は別。 本格的なAPIは価格と取引の配信を分け、口座状態を変える操作の制限が読取より大幅に低い場合があります。急変時に毎秒十数回ストップ変更するコードは、取引経路で遮断され、開いたポジションに一時アクセスできなくなるかもしれません。ブローカーの寛容さを期待せず、自分で変更頻度を制限する理由です。

8. ノートブックからサービスへ：執行コードとスクリプトの違い

ここが本当の境界です。日誌統計のノートブックが途中で落ちても、再実行すれば済みます。口座に触れるコードは同じ誤りでも、開いたポジション、途中の注文変更、二重エントリーを残します。品質の違いではなく、一方には取り消せない結果があるという違いです。

自動化の三段階 – 技術ではなく、最後に誰が決めるかが違います。

段階1：通知。 条件を確認してメッセージを送り、口座アクセスや発注トークンは不要です。判断と執行は人間です。コードの誤りが苛立ちだけで済むのはこの段階だけです。

段階2：準備と承認。 銘柄、方向、数量を組み立て、操作者の明示的承認後に送ります。計算ミスをポジションになる前に発見し、人を判断に残します。保守を行う意図のない個人には、通常は中間段階でなく到達点です。ただし研究結果ではなく編集上の判断です。

段階3：承認なしの稼働。 自動で新規・決済します。上の全要素、練習口座、その後最小数量口座での試験、初期稼働中の監督が必要です。次へ進むまでの取引数は切りのよい数でなく標本サイズと検出力の問題であり、 テストの記事で扱います。

安全ゲート付きの執行層

発注コードを示すなら、止める仕組みも示します。次のモジュールは全体を実行でき、通信を一切しません。注文を作り、制限を確認し、ゲートで止まります。送信は別の部分で、意図的にコードではありません。

"""執行層：検証、テストモード、本番前の明示ゲート。"""
import logging
import os
import sys
import time
import uuid

ALLOWED_INSTRUMENTS = {"EUR_USD"}
MAX_UNITS = 1_000          # 基軸通貨単位でありロットではない。表記はAPIによる

# 本番確認値はこのファイルにはないので読み取れない。
# プログラムは環境変数から読み、正しい値として、
# 操作者が意識して取引する口座番号を受け取る。起動前にシェルへ
# 手入力する。この障壁の範囲は、試験出力の下の囲みに記す。
CONFIRMATION_VARIABLE = "BROKER_LIVE_CONFIRMATION"

# この設定がないとlog.infoは残らない。既定のrootレベルは、
# WARNINGで出力先もない。ファイルは障害後の痕跡を残し、
# ストリームは同じ内容を即時表示する。大規模プログラムではログ設定は、
# モジュールでなくプログラム入口に置く。
# 監査時刻はUTCと明示的なZで記録する。
logging.Formatter.converter = time.gmtime
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s.%(msecs)03dZ %(levelname)s %(name)s %(message)s",
    datefmt="%Y-%m-%dT%H:%M:%S",
    handlers=[logging.FileHandler("execution.log", encoding="utf-8"),
              logging.StreamHandler(sys.stdout)],
)

log = logging.getLogger("execution")


class SafetyGateError(RuntimeError):
    """注文がプロセスを出る前に止める。"""


def new_order_id() -> str:
    """発注意図につき一度だけ呼び、送信前に保存する。

    同じ注文の再試行では必ず同じ値を使う。
    そうしなければ重複を認識できない。
    """
    return str(uuid.uuid4())


def build_order(instrument: str, units: int, order_id: str) -> dict:
    if instrument not in ALLOWED_INSTRUMENTS:
        raise SafetyGateError(f"未許可の銘柄です：{instrument}")
    # boolはintの派生なのでisinstance(True, int)はtrueになる。明示的に
    # 拒否しなければ真偽値が検証を通りpayloadに入る。
    if isinstance(units, bool) or not isinstance(units, int) or units == 0:
        raise SafetyGateError("数量はゼロ以外の整数にしてください")
    if abs(units) > MAX_UNITS:
        raise SafetyGateError(
            f"数量{abs(units)}が上限を超えました。上限：{MAX_UNITS}")
    return {
        "order": {
            "type": "MARKET",
            "instrument": instrument,
            "units": str(units),
            # 識別子は外部由来で、試行間で変わらない。
            # ここで作れば再試行ごとに別値になってしまう。
            # フィールド名と制約は該当APIによる。
            "clientExtensions": {"id": order_id},
        }
    }


def _operator_confirmation() -> str:
    """ファイル外の値を読む。コードはその値を知らず、どこにも持たない。"""
    return os.environ.get(CONFIRMATION_VARIABLE, "").strip()


def submit_order(session, cfg, instrument: str, units: int, order_id: str,
                  test_mode: bool = True) -> dict | None:
    """標準では何も送らない。本番ホストには二つの判断が必要。"""
    order = build_order(instrument, units, order_id)

    if test_mode:
        log.info("テストモード[%s] 注文を構築。送信なし：%s",
                 cfg.environment, order)
        return None

    if cfg.environment == "live" and _operator_confirmation() != cfg.account:
        raise SafetyGateError(
            "本番ホストには操作者の明示的承認が必要です")

    # ゲートを通過したところでモジュールは終わる。送信はない。
    # 検証可能な形で実行できないためで、構造は次の概要で説明する。
    raise NotImplementedError(
        "このモジュールは送信未実装です。構造例を参照してください")

注文IDは試行ごとでなく一度作る。 切断後に重複を認識するには、値を、 発注の意図が生まれた時点で作り、送信前に永続保存し、同じ注文の再試行で再利用します。もし、 uuid.uuid4() が注文構築内にあれば、毎回別値を送り、二回目を最初の反復と判断できません。そのため、 build_order() は自分で生成せず、引数で受け取ります。

IDだけでは解決しないこと。 上の、 clientExtensions は一社の方式で、業界標準でも文書化された冪等性保証でもありません。OANDAは任意の独自データと説明し、「MT4に関連付けた口座ではtradeClientExtensionsを設定・変更・削除しない」と別途警告します ^[14] （確認日2026-08-22）。同じIDの成行を重複拒否する約束はありません。成行は即時約定するため検出の時間窓も最小です。重複判定はフィールドでなく、送信の、 前 に書く永続的な意図台帳と、障害後に取得した注文・取引一覧との照合に依存します。自分のAPIのフィールド名と制約を確認してください。他社では名称が違うか、存在しません。

注文送信：構造の概要

次はコピー用関数ではなく、実装に移せる操作順序の構造例で、そのまま実行するものではありません。省略ミスではなく意図的な形式です。編集側はこの分岐を実行しておらず、実行する予定もありません。ブローカー口座が必要で、他人の口座で取引コードを試すのはテストでなく他人のリスクだからです。完全関数なら、誰も実行していない成果物を載せ、未検証コードを信じることを教えてしまいます。同じ判断は、 記事11.5 のMQLと、 記事11.2のPine Scriptにも適用しました。

構造の概要 - 注文送信と応答欠落への対処
HTTP層の方向性を示す記述。完全な実行可能コードではない。

宣言: StanNieznany - 応答がなく、注文の存在を確認できない。
             通信エラーの派生でない独立した型。通信エラーは
             再試行できるが、この状態では絶対に再試行しない。

入力: 構築・検証済み注文、order_id、設定、HTTPセッション
出力: ブローカーの応答かStanNieznany。黙った成功は禁止。

1. 永続台帳 - 何より先に
   プロセスメモリ外へ書く: order_id、UTC時刻、
   instrument, units, environment, status := INTENT
   次へ進む前にディスクへの書込を確定する
   なければ再起動で意図を失い、
   起動後にすでに試した操作が分からなくなる

2. ログ - 送信より先に
   記録: order_id + environment + 注文内容
   トークンは絶対に記録しない
   口座番号は末尾四文字以外を隠す
   ログは事後に人が読み、段階1の台帳はプログラムが読む

3. 一要求、再試行ゼロ
   API資料に示す注文経路へPOST
   認証ヘッダーのトークンは設定から取得し、コードには置かない
   タイムアウト必須。なければ無期限に待つ場合がある

4. 応答が来ない（タイムアウトまたは通信障害）
   再試行禁止。POSTは冪等でない
   台帳のstatus := UNKNOWN
   StanNieznanyを発生
   状態照合: 直近数分の注文と取引をブローカーに問い合わせ、
   order_idを探す。注文の存在を
   確定して初めて新しい判断を行う
   （この手順自体は概要に含めない。直後の段落を参照）

5. 応答が来た
   記録: HTTPコード + 切り詰めた応答本文
   コードが(200, 201)外 -> status := REJECTED -> エラー
   本文がJSONでない   -> status := UNKNOWN  -> StanNieznanyを発生
   それ以外            -> status := ACCEPTED -> 応答を返す

この概要に含めない、含められないもの:
   貼り付けて口座に対して実行できる完成関数はない。
   保護水準フィールド名もない。それは他者の例でなく
   特定APIの資料から取得するものだからである。

この概要が行わないこと も、記載したことと同じく重要です。ストップロスも利益確定も追加しません。APIごとに別フィールド、距離か絶対価格、有効期限が異なるため、他人の例ではなく自分のブローカー資料から実装する必要があります。実務上の影響は第4段階です。 StanNieznany の後は、保護されず、存在するかも不明なポジションを持つ可能性があります。送信できるコードでは保護水準をエントリーと同じ要求に含めます。そうしないとストップのない時間窓が生まれます。

欠けるのは保護水準だけではありません。「研究ノートブック ≠ 執行プログラム」の一覧のうち、上の実行可能モジュールにも概要にも監視、キルスイッチ、配備手順はありません。再起動後の状態保持は概要の第1段階に要件だけがあり、実装はありません。ログ・状態・テストの節はそれらを説明するだけでコードを示しません。結論は同じです。構造を学ぶ資料であり、執行プログラムではありません。

安全ゲートはオフラインで試験済み です。通信、ブローカー、秘密なしの模擬設定を使いました。出力だけでなく全テストを示します。なければ、この記事が警戒する、言葉だけを信じる行為を求めることになるからです。

"""通信なしのゲート試験：模擬設定、秘密なし、POSTなし。"""
from dataclasses import dataclass


@dataclass(frozen=True)
class MockConfiguration:
    environment: str = "practice"
    host: str = "https://api-fxpractice.oanda.com"
    token: str = "dummy-token-for-offline-test"
    account: str = "000-000-0000000-000"


def test_gates() -> None:
    cfg = MockConfiguration()
    order_id = new_order_id()
    print("設定：", cfg.environment, cfg.host, "token_length=", len(cfg.token))

    # session=Noneは意図的。もしPOSTを試みても、
    # 黙って通らずAttributeErrorで失敗する。
    result = submit_order(None, cfg, "EUR_USD", 100, order_id)
    assert result is None
    print("1. テストモード：送信なし - OK")

    for description, instrument, units in [
        ("未許可の銘柄", "GBP_JPY", 100),
        ("上限超過の数量", "EUR_USD", 5000),
        ("数量ゼロ", "EUR_USD", 0),
        ("数値の代わりに真偽値", "EUR_USD", True),
    ]:
        try:
            build_order(instrument, units, order_id)
        except SafetyGateError as error:
            print(f"2. ゲート［{description}］：", error)
        else:
            raise AssertionError(f"想定ゲートが作動しませんでした：{description}")

    cfg_live = MockConfiguration(environment="live", host="https://api-fxtrade.oanda.com")
    try:
        submit_order(None, cfg_live, "EUR_USD", 100, order_id, test_mode=False)
    except SafetyGateError as error:
        print("3. ゲート［未承認の本番］：", error)
    else:
        raise AssertionError("本番が承認なしで通過しました")

    print("全ゲートを確認済み")


if __name__ == "__main__":
    test_gates()

実行結果は第2節と同じPython 3.12.10、通信なし、標準外パッケージなしです。送信を外したので、執行モジュールは、 requestsを必要としません。二行目はログ記録で、同じ内容がファイル、 execution.log と標準出力に同時に流れます。時刻と注文IDは毎回変わりますが、残りは文字単位で再現します。

設定： practice https://api-fxpractice.oanda.com token_length= 28
2026-11-02T12:00:00.011Z INFO execution テストモード[practice] 注文を構築。送信なし：{'order': {'type': 'MARKET', 'instrument': 'EUR_USD', 'units': '100', 'clientExtensions': {'id': '4ae82fd8-2de4-4117-97fc-c4aac3d11d99'}}}
1. テストモード：送信なし - OK
2. ゲート［未許可の銘柄］： 未許可の銘柄です：GBP_JPY
2. ゲート［上限超過の数量］： 数量5000が上限を超えました。上限：1000
2. ゲート［数量ゼロ］： 数量はゼロ以外の整数にしてください
2. ゲート［数値の代わりに真偽値］： 数量はゼロ以外の整数にしてください
3. ゲート［未承認の本番］： 本番ホストには操作者の明示的承認が必要です
全ゲートを確認済み

この出力で確認できること、できないこと。 未許可銘柄、超過数量、ゼロ、真偽値、未承認の本番という五つの負例で、想定ゲートが動きました。assertでテストモードも確認し、監査では±1000の境界、超過、不正型も確認しました。最初の引数は、 Noneで送信も未実装なので、発注はできません。API、ログの耐障害性、全機密フィールドのマスクは試験しません。ダミートークンは長さだけ表示します。自分のプロジェクトでは、ログの書込・復旧、伏字規則、ディスク障害を別途確認します。

実APIに対する挙動は確認できません。このモジュールには送信分岐自体がなく、構造例に置き換えたからです。「実行していない」と「実行対象がない」の違いです。前者は貼り付け可能な成果物を残し、後者は自分で実装し、その過程で自社資料を読む必要のある構造を残します。

安全ゲートの欠落も検出する必要がある。 そのためテストモードの結果をassertし、負例で期待する例外が出なければ、 AssertionError を起こします。最終メッセージは全確認後だけ出ます。これは小さな例で、応答形式、全型、状態復旧を網羅しません。大きなプロジェクトでは、 pytestに移し、正例を加え、検証処理の変更時に毎回実行します。

通貨単位はロットではない。 横の、 MAX_UNITS というコメントは飾りではありません。APIは基軸通貨単位、ロット、契約数など異なる数量表記を使います。誤っても構文エラーや例外は出ず、正しく動くコードが意図の二、三桁上のリスクを持ちます。送信前に資料を確認し、練習口座の最小取引で確かめます。検証処理には二つの限界もあります。pandasやNumPyで計算した数量は、 numpy.int64型で、組込の、 int の派生ではありません。整数なのに整数条件で拒否されます。安全側の失敗ですが紛らわしいので、DataFrameの値を、 int(...) に通してから、 build_order()へ渡します。第二に、 order_id は一切検査せず、型注釈も強制しません。重複認識の鍵なので、API資料の許容長と形式を確認し、数量制限と同じく検査します。

ログ、状態、テスト

ログに置き換える： print(). 標準の、 logging モジュールは重要度、時刻、ファイル出力を提供しますが、自動では有効になりません。設定しなければ既定は、 WARNING で、出力先もないため、 log.info(...) は黙って捨てられます。記録しているように見えて何も残らず、障害後に空のログを発見する原因です。設定は実行スクリプトの入口で一度だけ行い、インポートするモジュールでは行いません。上の、 logging.basicConfig(...) は、モジュール自体が全プログラムだから置いています。コメントでも大規模プロジェクトでは不適切な場所と明記しました。モジュール内設定はインポート時に全プロセスのログを支配し、本当の入口で後から、 logging.basicConfig(...) しても何も起きません。root loggerに出力先があれば無効で、例外は、 force=Trueを付ける場合です。しかもエラーを通知しません ^[4]。自分の設定が消え、取込先が指定した場所へ記録されます。同じ無言の欠陥の別経路です。判断と根拠、要求内容、応答を記録し、HTMLエラーで埋まらないよう応答を切り詰め、診断時に上限を増やします。トークンだけは絶対に書きません。応答のないログでは、原因が論理、通信、サーバーのどこかを後で判断できません。 タイムスタンプはlogging.Formatter.converter = time.gmtimeと明示的なZによりUTCで記録します。この設定がなければ、通常はシステムのローカル時刻になります。ブローカーとシステムのログは同じUTC基準で照合し、別のゾーンは表示にだけ使ってください。

状態は再起動後も残す。 プロセスはいずれ更新、停電、自分の、 Ctrl+Cで止まります。送信注文がメモリにしかなければ、再起動後に実施済みのことが分かりません。最小の対策は送信前にIDをファイルかDBに保存し、起動時にAPIの口座状態と照合することです。それが切断後の二重エントリーを防ぎます。

通信なしのテスト。 シグナル論理と制限検証は、上の安全ゲート同様、準備したデータで接続せず試せます。稼働APIが必要なら単体テストではなく、変更のたびに実行できません。まず、エントリー条件、数量計算、空データと欠落データの挙動など、オフラインで確認できるものから始めます。

制御された停止。 終了信号には、現在処理を終え、状態を保存し、終了するよう応答します。発注途中で切断されないためです。別に、日次損失上限で新規を止める条件も有用です。上限と計算法は、 リスク管理の章で説明します。コードではライブラリ機能でなく、毎回のエントリー前の通常の条件です。

関連項目： アルゴリズム取引 – システム構造と意思決定全体での自動化の位置。

9. Pythonコードのルックアヘッドとデータ漏洩

ルックアヘッドは方法論でなく、一行のコードの誤りです。そのため方法論記事ではなくここで扱います。ある時点に割り当てた結果に、その時点にはない情報を使うことです。

pandasの典型は、同じ行のシグナル列とリターン列を掛けるだけです。行、 t のリターンは前の足の終値から、足、 tの終値までに生じます。一方、足、 t の終値から計算したシグナルは、そのリターンが生じた後に分かります。掛けると、判断時に知らなかった値動きの利益を自分に与えます。

次の例の前提： 合成した八つの毎時観測、UTC索引、例示値、コストなしです。目的はルール評価でなくずれの仕組みです。環境は第2節と同じPython 3.12.10、pandas 3.0.5です。

"""一行のルックアヘッド：同じルール、異なる二つの結果。"""
import pandas as pd

index = pd.date_range("2026-06-01", periods=8, freq="1h", tz="UTC")
df = pd.DataFrame(
    {"close": [1.0800, 1.0810, 1.0825, 1.0815, 1.0830, 1.0845, 1.0840, 1.0855]},
    index=index,
)
df["sma"] = df["close"].rolling(3).mean()
df["return"] = df["close"].pct_change()

# ルール：終値が三本平均より上なら買いポジション。
df["position"] = (df["close"] > df["sma"]).astype(int)

# 誤り：行tの位置を同じ行のリターンに掛ける。
# リターンtはt-1終値からt終値までに生じるので、判断は、
# エントリー時点に存在しない価格を使ってしまう。
df["result_with_lookahead"] = df["position"] * df["return"]

# 正しい：t終値で決めた位置が得るのはt+1のリターン。
df["result_without_lookahead"] = df["position"].shift(1) * df["return"]

print(df[["result_with_lookahead", "result_without_lookahead"]].sum().round(6))

実行結果：

result_with_lookahead       0.005082
result_without_lookahead    0.001384
dtype: float64

解釈。 同じルール、同じデータで、一回の、 shift(1) だけの違いにより、単純な各期リターンの合計が0.005082から0.001384、元の30パーセント未満になります。八観測なので測定でなく例示ですが、数年なら普通の戦略を華々しく、華々しい戦略を実現不能に見せられます。どちらも無次元、スプレッド・手数料なしで、控除後の正しい版はさらに下がります。

計算自体への留保。 pct_change() は単純リターンなので、第4節の通り加算でなく積で複利計算します。合計できるのは対数リターンだけです。ここで足すのは意図した近似です。千分の一程度では、正しい複利との差はこの例で0.00001未満で、示したい差より二桁以上小さいからです。ずれの仕組みでなく口座損益を示すなら、 (1 + r) を掛け合わせるか、対数リターンに変えて合計します。

二重シフトに注意。 位置列か入力データのどちらか一箇所だけをずらします。両方は不可です。価格をずらしてシグナルを作り、さらに位置をずらすと二期間遅れ、黙って結果を過小評価します。ルックアヘッドほど危険ではなくても、同じく誤りです。

Pythonでの他のルックアヘッド – すべて「判断時にこの値は分かっていたか」に帰着します。

• 確定後にしか分からない当該足の高値・安値をエントリー条件に使う
• 訓練・検証分割前に全データで尺度調整や正規化を行い、未来情報を持ち込む
• 後の足で平滑化や転換点表示を行い、過去値と実際に見えた値が異なる指標
• 当初公表値でなく改定後のマクロデータ
• シグナルが生まれた足の終値での約定

データ準備時の漏洩。 第二項目はPythonで反射的に書きがちなので、補足が必要です。分割前にDataFrame全体でスケーラーを呼ぶ一行は無害に見えても、未来の分布を検証部分へ持ち込みます。変換パラメータは過去だけで決め、後のデータに適用します。逆ではありません。

多重検定。 同じデータで試す設定が多いほど、最高成績が偶然の可能性は高まります。明示した帰無モデルで10,000試験それぞれの偽陽性確率が0.05なら、「有意」とされる期待数は0.05 × 10,000 = 500です。確率の和による期待値なので独立性は不要で、依存はこの前提下で平均でなく件数分布を変えます。この規模から、結果と試行数を示す必要が分かります。

試行数を補正するDeflated Sharpe Ratioは、BaileyとLópez de Pradoが2014年に提案しました ^[12]。既成実装を使う前に、ライブラリの状態とライセンスを確認します。配布方式が変わったものもあります。式は元論文にあり、自分で実装もできます。

この節で扱わないこと。 過剰適合、標本選択、ウォークフォワード、生存者バイアス、コスト増加への頑健性は方法論なので、 バックテストとフォワードテストの記事で扱います。ここではコードで壊れ、コードで確認できるものだけを扱います。

成績を信じる前のコード確認：

1. 行、 t の結果は、判断前に利用できる情報だけを使うか
2. シフトはシグナルかデータの一方に一度だけか
3. 約定時点と価格を暗黙にせず、明示したか
4. コストを含むか、意識して省き明記したか
5. 訓練・検証分割は時系列順か
6. 変換は過去部分だけで決めたか
7. 試した設定数を結果と記録したか
8. 時刻はゾーン付きで、重複なく昇順か

10. 本当に必要なコード量

開発者になる必要はありません。自分のコードやダウンロードした他人のスクリプトに欺かれない程度の知識が必要です。

最終列は概算で、開始時の知識と週の実作業時間次第です。約束でなく規模の目安です。

段階1と2は市場の優位性を与えない。 得られるのは他者の解決策の妥当性を確認し、検証不能なシステムを買わない能力です。実際の価値ですが、販売者の訴求とは種類が違います。段階4は執行ソフト保守に必要なのであって、取引を上達させるためではありません。金融用途の概観はHilpischの教科書にあります ^[11]。ただしすべてのコード書籍同様、公開後に一部APIが変わるので現行版で確認してから使います。

合理的な学習順序 – 前段を土台にし、動くものを完成させます。

1. 言語の基礎 – 変数、リスト、関数、条件。推測せず他人のコードを読める程度です
2. 自分のデータでpandas – 日誌を読み、曜日・銘柄別統計を求めます。資料より本なら、pandas作者McKinneyの本がこの段階を網羅します ^[10]
3. グラフ – 前段の計算から資産曲線とドローダウンを描きます
4. データ品質 – 第4節を自分のファイルで再現。ゾーン、重複、欠落です
5. 読取専用の接続 – 提供されるなら発注権限のないトークンで、練習口座APIから取得します

順序は意図的です。取得データを検査できて初めて第五段階です。さもなければ作業でなく誤りを自動化します。

一つの作業から始めます。自分のデータを読み、三行なら手で確認できる計算を比較します。指折り計算と合えば基礎ができています。違えば最初のバグを発見でき、理解しないまま即動くスクリプトよりよい出発点です。

Pythonは悪い戦略を直しません。悪かったと早く示すのが最善です。