PythonのloggingでトレードシグナルとMLを自動記録する方法

自動化・運用

※本記事のコードや情報は執筆時点の仕様に基づいています。投資は自己責任であり、必ずデモ環境や少額資金でテストした上で運用してください。

Pythonで株価分析や自動売買のスクリプトを運用し始めると、「いつシグナルが発生したのか」「スクリプトは正常に動作したのか」「どこでエラーが起きたのか」を後から確認したい場面が必ず訪れます。

📘 外部参考Python 公式Python 公式ドキュメント(日本語)

print() 文でコンソールに出力するだけでは、ターミナルを閉じた瞬間にすべての情報が消失します。これはトレードの記録としても、デバッグの手段としても致命的です。

この問題を解決するのが、Python標準ライブラリのloggingモジュールです。この記事では、loggingの基本概念から、シグナル判定ログをファイルに自動保存する実践的な設定例、さらにログローテーション(自動分割)の実装まで、コピペで即座に使えるコードとともに解説します。

📘 外部参考logging — Python公式ドキュメント(日本語)Logging HOWTO(公式)

なぜprint()ではなくloggingを使うべきなのか

まずは print() とloggingモジュールの違いを明確にしておきます。

print()の限界

print() には以下の問題点があります。

  • 記録が残らない: コンソール出力のみで、ファイルに保存されない
  • 重要度の区別ができない: エラーも通常メッセージも同じ見た目で出力される
  • 本番運用で邪魔になる: 不要なprint文を削除し忘れると出力が汚れる
  • タイムスタンプがない: いつ出力されたか記録されない

loggingモジュールのメリット

機能 print() logging
ファイル保存 ×(追加実装が必要) ◎(標準機能)
重要度レベルの分類 × ◎(DEBUG〜CRITICALの5段階)
タイムスタンプの自動付与 ×
出力先の切り替え × ◎(コンソール・ファイル・メール等)
本番/開発の切り替え × ◎(レベル変更で制御可能)
外部ライブラリ不要 ◎(Python標準ライブラリ)

loggingモジュールはPythonに標準搭載されており、pip install は不要です。追加コストなしで、プロフェッショナルなログ管理が実現できます。

loggingの基本概念:5つのログレベル

loggingモジュールでは、メッセージの重要度を5段階のレベルで分類します。

ログレベル一覧と使い分け

レベル 数値 用途 トレードシステムでの具体例
DEBUG 10 開発時の詳細情報 取得した株価データの中身を確認
INFO 20 正常動作の記録 「買いシグナル発生」「データ取得完了」
WARNING 30 注意が必要な状況 「データ取得に遅延あり」「出来高が異常に少ない」
ERROR 40 エラー発生(処理は継続) 「特定銘柄のデータ取得に失敗」
CRITICAL 50 致命的エラー(処理続行不可) 「API接続が完全に断絶」

ログレベルにはしきい値の概念があります。例えばレベルをINFOに設定すると、INFO以上(INFO・WARNING・ERROR・CRITICAL)のメッセージのみが出力され、DEBUGは無視されます。

開発時と本番運用時のレベル設定

  • 開発・テスト時: DEBUG(すべてのメッセージを出力)
  • 本番運用時: INFO または WARNING(必要な情報のみ出力)

この切り替えを1行のコード変更で実現できるのが、loggingモジュールの大きな利点です。

【コピペOK】基本的なlogging設定コード

まずはloggingの最もシンプルな設定から始めます。

最小構成のlogging設定

以下のコードを logging_basic.py として保存・実行してください。


import logging

# ==============================
# 基本的なlogging設定
# ==============================
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
)

logger = logging.getLogger(__name__)

# ==============================
# 各レベルのログ出力テスト
# ==============================
def main():
    logger.debug("これはDEBUGメッセージです(INFO設定では表示されません)")
    logger.info("スクリプトを開始しました")
    logger.warning("データ取得に通常より時間がかかっています")
    logger.error("銘柄コード 9999.T のデータ取得に失敗しました")
    logger.critical("API接続が完全に断絶しました")

if __name__ == "__main__":
    main()

実行結果:


2026-02-20 17:30:00 [INFO] スクリプトを開始しました
2026-02-20 17:30:00 [WARNING] データ取得に通常より時間がかかっています
2026-02-20 17:30:00 [ERROR] 銘柄コード 9999.T のデータ取得に失敗しました
2026-02-20 17:30:00 [CRITICAL] API接続が完全に断絶しました

level=logging.INFO と設定しているため、DEBUGメッセージは出力されていません。

format文字列の主要パラメータ

format 引数で出力形式をカスタマイズできます。主要なパラメータは以下のとおりです。

  • %(asctime)s — タイムスタンプ
  • %(levelname)s — ログレベル名(INFO、ERRORなど)
  • %(message)s — ログメッセージ本文
  • %(filename)s — ファイル名
  • %(lineno)d — 行番号
  • %(funcName)s — 関数名

デバッグ時には %(filename)s:%(lineno)d を追加すると、どのファイルの何行目でログが出力されたか一目で分かるようになります。

【コピペOK】ログをファイルに自動保存する設定

コンソール出力だけではなく、ファイルにログを保存する設定を実装します。

コンソールとファイルの同時出力

以下のコードは、loggingのHandler機能を使ってコンソールとファイルの両方にログを出力する構成です。logging_file.py として保存してください。


import logging
import os
from datetime import datetime

# ==============================
# ログ出力設定(コンソール+ファイル)
# ==============================
def setup_logger(log_dir="logs"):
    # ログ保存用ディレクトリの作成
    os.makedirs(log_dir, exist_ok=True)

    # ログファイル名に日付を含める
    log_filename = datetime.now().strftime("%Y%m%d") + "_trade.log"
    log_filepath = os.path.join(log_dir, log_filename)

    # ロガーの取得
    logger = logging.getLogger("TradeLogger")
    logger.setLevel(logging.DEBUG)

    # 既存のハンドラをクリア(重複防止)
    if logger.handlers:
        logger.handlers.clear()

    # フォーマッタの定義
    formatter = logging.Formatter(
        fmt="%(asctime)s [%(levelname)-8s] %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S"
    )

    # ファイルハンドラ(DEBUG以上をすべて記録)
    file_handler = logging.FileHandler(
        log_filepath, encoding="utf-8"
    )
    file_handler.setLevel(logging.DEBUG)
    file_handler.setFormatter(formatter)

    # コンソールハンドラ(INFO以上のみ表示)
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.INFO)
    console_handler.setFormatter(formatter)

    # ハンドラの登録
    logger.addHandler(file_handler)
    logger.addHandler(console_handler)

    logger.info(f"ログファイル: {log_filepath}")
    return logger


# ==============================
# 動作テスト
# ==============================
def main():
    logger = setup_logger()

    logger.debug("デバッグ情報:ファイルにのみ記録されます")
    logger.info("スクリプト開始")
    logger.info("データ取得処理を実行中...")
    logger.warning("出来高が通常の50%以下です")
    logger.error("銘柄 6758.T のデータ取得に失敗")
    logger.info("スクリプト終了")

if __name__ == "__main__":
    main()

このコードのポイント:

  • ログファイルは logs/ フォルダに日付付きで自動生成される
  • ファイルにはDEBUG以上すべてが記録される
  • コンソールにはINFO以上のみが表示される
  • ファイルのエンコーディングはUTF-8(日本語対応)

出力されるログファイルの例

logs/20260220_trade.log の中身は以下のようになります。


2026-02-20 17:30:00 [INFO    ] ログファイル: logs/20260220_trade.log
2026-02-20 17:30:00 [DEBUG   ] デバッグ情報:ファイルにのみ記録されます
2026-02-20 17:30:00 [INFO    ] スクリプト開始
2026-02-20 17:30:00 [INFO    ] データ取得処理を実行中...
2026-02-20 17:30:00 [WARNING ] 出来高が通常の50%以下です
2026-02-20 17:30:00 [ERROR   ] 銘柄 6758.T のデータ取得に失敗
2026-02-20 17:30:00 [INFO    ] スクリプト終了

コンソールには表示されなかったDEBUGメッセージが、ファイルにはしっかり記録されています。これにより、通常はクリーンな表示を維持しつつ、問題発生時にはファイルで詳細を追跡できます。

【コピペOK】シグナル判定ログを記録する実践コード

ここからは、実際の株価分析シナリオを想定した実践的なログ記録コードを実装します。yfinanceで株価データを取得し、移動平均線のゴールデンクロスを検出してログに記録する構成です。

📘 外部参考yfinance 公式GitHub

シグナル判定スクリプト全体

以下のコードを signal_logger.py として保存してください。


import logging
import os
from datetime import datetime

import yfinance as yf
import pandas as pd

# ==============================
# ログ設定
# ==============================
def setup_logger(log_dir="logs"):
    os.makedirs(log_dir, exist_ok=True)
    log_filename = datetime.now().strftime("%Y%m%d") + "_signal.log"
    log_filepath = os.path.join(log_dir, log_filename)

    logger = logging.getLogger("SignalLogger")
    logger.setLevel(logging.DEBUG)

    if logger.handlers:
        logger.handlers.clear()

    formatter = logging.Formatter(
        fmt="%(asctime)s [%(levelname)-8s] %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S"
    )

    file_handler = logging.FileHandler(log_filepath, encoding="utf-8")
    file_handler.setLevel(logging.DEBUG)
    file_handler.setFormatter(formatter)

    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.INFO)
    console_handler.setFormatter(formatter)

    logger.addHandler(file_handler)
    logger.addHandler(console_handler)
    return logger


# ==============================
# 株価データ取得
# ==============================
def fetch_stock_data(symbol, period="6mo", logger=None):
    logger.info(f"{symbol} のデータを取得中(期間: {period})")

    try:
        ticker = yf.Ticker(symbol)
        df = ticker.history(period=period)

        if df.empty:
            logger.error(f"{symbol} のデータが空です")
            return None

        logger.info(f"{symbol} のデータ取得完了({len(df)}件)")
        logger.debug(f"期間: {df.index[0].date()} 〜 {df.index[-1].date()}")
        return df

    except Exception as e:
        logger.error(f"{symbol} のデータ取得中にエラー発生: {e}")
        return None


# ==============================
# シグナル判定(ゴールデンクロス / デッドクロス)
# ==============================
def check_signals(df, symbol, short_window=5, long_window=25, logger=None):
    logger.info(f"{symbol} のシグナル判定開始(短期={short_window}日, 長期={long_window}日)")

    df = df.copy()
    df["SMA_short"] = df["Close"].rolling(window=short_window).mean()
    df["SMA_long"] = df["Close"].rolling(window=long_window).mean()
    df.dropna(inplace=True)

    if len(df) < 2:
        logger.warning(f"{symbol}: 判定に必要なデータが不足しています")
        return

    # 直近2日分でクロス判定
    prev = df.iloc[-2]
    curr = df.iloc[-1]

    prev_diff = prev["SMA_short"] - prev["SMA_long"]
    curr_diff = curr["SMA_short"] - curr["SMA_long"]

    close_price = curr["Close"]
    sma_short = curr["SMA_short"]
    sma_long = curr["SMA_long"]

    logger.debug(
        f"{symbol} | 終値: {close_price:.1f} | "
        f"SMA{short_window}: {sma_short:.1f} | "
        f"SMA{long_window}: {sma_long:.1f}"
    )

    if prev_diff <= 0 and curr_diff > 0:
        logger.info(f"★ {symbol} 【買いシグナル】ゴールデンクロス検出 "
                     f"(終値: {close_price:.1f})")
    elif prev_diff >= 0 and curr_diff < 0:
        logger.info(f"★ {symbol} 【売りシグナル】デッドクロス検出 "
                     f"(終値: {close_price:.1f})")
    else:
        logger.info(f"  {symbol} シグナルなし(終値: {close_price:.1f})")


# ==============================
# メイン処理
# ==============================
WATCHLIST = ["7203.T", "6758.T", "9984.T", "8306.T"]

def main():
    logger = setup_logger()
    logger.info("=" * 50)
    logger.info("シグナル判定スクリプト開始")
    logger.info(f"監視銘柄: {WATCHLIST}")
    logger.info("=" * 50)

    for symbol in WATCHLIST:
        df = fetch_stock_data(symbol, logger=logger)
        if df is not None:
            check_signals(df, symbol, logger=logger)

    logger.info("=" * 50)
    logger.info("シグナル判定スクリプト終了")
    logger.info("=" * 50)

if __name__ == "__main__":
    main()

コードの処理フロー

  1. ログ設定を初期化(ファイル+コンソールの二重出力)
  2. ウォッチリストの各銘柄について、yfinanceで6ヶ月分の株価データを取得
  3. 5日移動平均線と25日移動平均線を計算
  4. 直近2日間のクロス状態を比較し、ゴールデンクロスまたはデッドクロスを検出
  5. 判定結果をINFOレベルでログに記録

ログローテーション(自動分割)の実装

ログファイルが際限なく肥大化するのを防ぐために、ログローテーションを設定します。

TimedRotatingFileHandlerの活用

Python標準の logging.handlers モジュールに含まれる TimedRotatingFileHandler を使えば、日付単位でログファイルを自動分割できます。


import logging
from logging.handlers import TimedRotatingFileHandler
import os

# ==============================
# ログローテーション設定
# ==============================
def setup_rotating_logger(log_dir="logs"):
    os.makedirs(log_dir, exist_ok=True)
    log_filepath = os.path.join(log_dir, "trade.log")

    logger = logging.getLogger("RotatingLogger")
    logger.setLevel(logging.DEBUG)

    if logger.handlers:
        logger.handlers.clear()

    formatter = logging.Formatter(
        fmt="%(asctime)s [%(levelname)-8s] %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S"
    )

    # 日次ローテーション(7日分保持)
    rotating_handler = TimedRotatingFileHandler(
        log_filepath,
        when="midnight",
        interval=1,
        backupCount=7,
        encoding="utf-8"
    )
    rotating_handler.suffix = "%Y%m%d"
    rotating_handler.setLevel(logging.DEBUG)
    rotating_handler.setFormatter(formatter)

    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.INFO)
    console_handler.setFormatter(formatter)

    logger.addHandler(rotating_handler)
    logger.addHandler(console_handler)
    return logger

ローテーション設定の主要パラメータ

パラメータ 設定値 意味
when "midnight" 毎日深夜0時にファイルを切り替え
interval 1 1日ごとにローテーション
backupCount 7 過去7日分のログを保持(それ以前は自動削除)
encoding "utf-8" 日本語ログに対応

backupCount=7 に設定すると、8日目以降の古いログファイルは自動的に削除されます。ディスク容量の圧迫を防ぐために、運用環境では必ず設定してください。

よくあるエラーと対処法

ログが二重に出力される

同じスクリプト内で setup_logger() を複数回呼び出すと、ハンドラが重複登録されてログが二重・三重に出力されます。

本記事のコードでは以下の処理で対策済みです。


if logger.handlers:
    logger.handlers.clear()

この処理により、関数を再度呼び出してもハンドラがリセットされ、重複を防止できます。

ログファイルに日本語が文字化けする

FileHandlerencoding パラメータが未指定の場合、Windows環境ではシステムデフォルトのエンコーディング(cp932)が使用され、一部の文字が化ける可能性があります。

必ず encoding="utf-8" を明示的に指定してください。

ログファイルが作成されない

logs/ フォルダが存在しない場合、FileHandlerがエラーを発生させます。本記事のコードでは os.makedirs(log_dir, exist_ok=True) で自動作成していますが、独自にコードを書く際はフォルダの事前作成を忘れないようにしてください。

例外情報をログに記録したい

try-except ブロック内で例外のスタックトレースをログに記録するには、logger.exception() を使用します。


try:
    result = 1 / 0
except Exception:
    logger.exception("予期しないエラーが発生しました")

logger.exception() はERRORレベルでメッセージを出力し、さらにスタックトレース(エラーの発生箇所の詳細)を自動的に付加します。

まとめ

Pythonのloggingモジュールを活用することで、トレードシステムの運用品質は大幅に向上します。本記事で解説した内容を整理すると、以下のとおりです。

  • print() ではなくloggingモジュールを使うことで、タイムスタンプ付きのログをファイルに自動保存できる
  • ログレベル(DEBUG〜CRITICAL)を使い分けることで、開発時と本番運用時の出力量を1行で切り替えられる
  • FileHandlerとStreamHandlerの組み合わせで、コンソールとファイルへの同時出力が実現できる
  • TimedRotatingFileHandlerを使えば、日付単位でのログ分割と古いファイルの自動削除ができる
  • 例外情報は logger.exception() でスタックトレースごと記録する

ログは「問題が起きた後」に価値を発揮します。シグナル判定やデータ取得の処理には、最初からloggingを組み込んでおくことを強く推奨します。次のステップとして、タスクスケジューラとの連携による定期自動実行に進めば、完全自動のシグナル監視システムが完成します。

📘 外部参考タスクスケジューラ(Microsoft 公式ドキュメント・日本語)

タイトルとURLをコピーしました