自動売買のログをSQLiteで一元管理する設計〜AIデバッグ・最適化を助ける方法

自動売買のログをSQLiteで一元管理する設計〜AIデバッグ・最適化を助ける方法
なぜSQLiteなのか
DB設計：3テーブル構成
LogManagerクラスの実装
使い方の例
pandasで分析
勝率・利益集計SQL
まとめ

自動売買のログをSQLiteで一元管理する設計〜AIデバッグ・最適化を助ける方法

カテゴリ: AI×自動売買 Week 1 Python実装・コード

AI×自動売買システムの改善には、「なぜシグナルが発生したか」「なぜ注文が失敗したか」を後から追えるログが不可欠です。本記事ではSQLiteを使ってトレードログ・シグナルログ・システムログを一元管理する設計を解説します。

なぜSQLiteなのか

Python標準ライブラリに内蔵（インストール不要）
ファイル1つで完結するデータベース
SQLでシグナル条件・勝率・利益率を簡単に集計可能
pandasとの相性が良い

DB設計：3テーブル構成

import sqlite3
from pathlib import Path
from datetime import datetime

DB_PATH = Path("logs/trading.db")
DB_PATH.parent.mkdir(exist_ok=True)

def init_db(db_path=DB_PATH):
    conn = sqlite3.connect(db_path)
    cur = conn.cursor()
    
    # トレードログテーブル
    cur.execute('''
        CREATE TABLE IF NOT EXISTS trades (
            id          INTEGER PRIMARY KEY AUTOINCREMENT,
            timestamp   TEXT NOT NULL,
            symbol      TEXT NOT NULL,
            side        TEXT NOT NULL,
            qty         REAL NOT NULL,
            price       REAL,
            status      TEXT,
            order_id    TEXT,
            strategy    TEXT,
            notes       TEXT
        )
    ''')
    
    # シグナルログテーブル
    cur.execute('''
        CREATE TABLE IF NOT EXISTS signals (
            id          INTEGER PRIMARY KEY AUTOINCREMENT,
            timestamp   TEXT NOT NULL,
            symbol      TEXT NOT NULL,
            signal      TEXT NOT NULL,
            score       REAL,
            model       TEXT,
            features    TEXT
        )
    ''')
    
    # システムログテーブル
    cur.execute('''
        CREATE TABLE IF NOT EXISTS system_logs (
            id        INTEGER PRIMARY KEY AUTOINCREMENT,
            timestamp TEXT NOT NULL,
            level     TEXT NOT NULL,
            message   TEXT NOT NULL,
            module    TEXT
        )
    ''')
    
    conn.commit()
    conn.close()
    print(f"DB初期化完了: {db_path}")

init_db()

LogManagerクラスの実装

import sqlite3
import json
from pathlib import Path
from datetime import datetime, timezone

class LogManager:
    def __init__(self, db_path="logs/trading.db"):
        self.db_path = db_path
    
    def _now(self):
        return datetime.now(timezone.utc).isoformat()
    
    def log_trade(self, symbol, side, qty, price=None, status="submitted",
                  order_id=None, strategy=None, notes=None):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                '''INSERT INTO trades 
                   (timestamp,symbol,side,qty,price,status,order_id,strategy,notes)
                   VALUES (?,?,?,?,?,?,?,?,?)''',
                (self._now(), symbol, side, qty, price, status, order_id, strategy, notes)
            )
    
    def log_signal(self, symbol, signal, score=None, model=None, features=None):
        features_json = json.dumps(features) if features else None
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                '''INSERT INTO signals
                   (timestamp,symbol,signal,score,model,features)
                   VALUES (?,?,?,?,?,?)''',
                (self._now(), symbol, signal, score, model, features_json)
            )
    
    def log_system(self, level, message, module=None):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                '''INSERT INTO system_logs (timestamp,level,message,module)
                   VALUES (?,?,?,?)''',
                (self._now(), level, message, module)
            )

使い方の例

logger = LogManager()

# シグナル記録
logger.log_signal("SPY", "BUY", score=0.82, model="llm-v1",
                  features={"rsi": 32, "trend": "up"})

# 注文記録
logger.log_trade("SPY", "buy", qty=1, price=523.45,
                 status="filled", order_id="abc123", strategy="llm-momentum")

# システムログ
logger.log_system("INFO", "ボット起動完了", module="main")

pandasで分析

import pandas as pd
import sqlite3

conn = sqlite3.connect("logs/trading.db")
df = pd.read_sql("SELECT * FROM trades ORDER BY timestamp DESC LIMIT 20", conn)
conn.close()
print(df[["timestamp","symbol","side","qty","price","status"]])

勝率・利益集計SQL

-- シンボル別勝率
SELECT symbol,
       SUM(CASE WHEN status='filled' THEN 1 ELSE 0 END) AS total,
       AVG(CASE WHEN status='filled' THEN 1.0 ELSE 0.0 END) * 100 AS win_rate_pct
FROM trades
GROUP BY symbol;

-- ストラテジー別平均スコア
SELECT t.strategy, AVG(s.score) AS avg_score
FROM trades t
JOIN signals s ON DATE(t.timestamp) = DATE(s.timestamp) AND t.symbol = s.symbol
GROUP BY t.strategy;

まとめ

SQLiteはPython標準込みでインストール不要
trades/signals/system_logsの3テーブル構成が基本
LogManagerクラスで全記録操作を一元化
pandas + SQLでAIデバッグ・最適化が嬧1步進む