npm install さようなら — FastAPI + htmx でトレーディングシステムを作る

このシリーズについて

ついにこのシリーズも 最終回 を迎えました! 第1回から読んでくださった方、途中から合流してくれた方、本当にありがとうございます。

このシリーズでは FastAPI + htmx + Plotly を使って、日経225先物・オプションのリアルタイムトレーディングダッシュボードをゼロから作ってきました。最終回となる今回は、バックテストエンジンの設計 から Docker での本番デプロイ まで、「開発したシステムを実際に使える状態にする」ための仕上げ作業を行っていきます。

前提: このシリーズの第1〜4回を読んでいることを前提とします。プロジェクト構成、FastAPI の lifespan パターン、DB設計、htmx によるUI構築の基礎が分かっている前提で進めます。

シリーズ目次

回	タイトル
1	環境構築とプロジェクト設計
2	htmx でSPA風ダッシュボードを作る
3	ダークテーマUIとコンポーネント設計
4	DB設計とAlembicマイグレーション
5	証券APIとの接続
6	トレーディングエンジン — 司令塔の設計
7	戦略とリスク管理
8	バックテストと本番デプロイ（本記事）

---

目次

1. バックテストエンジンの設計
2. ヒストリカルデータのロード
3. パラメータ最適化
4. Plotly による結果可視化
5. Docker 本番運用のベストプラクティス
6. Slack 通知で取引状況をリアルタイムに把握する
7. Basic認証で本番環境を守る
8. シリーズ全体のまとめ
9. 今後の発展 — ここから先は読者の皆さんの番です

---

1. バックテストエンジンの設計

戦略を本番に投入する前に、過去のデータで検証する。これがバックテストです。「理論上は儲かるはず」を「過去データでは実際に儲かった」に変える、最も重要なステップですね。

第7回で設計したトレンドフォロー戦略（SMAクロスオーバー + RSI + MACD フィルタ）を、ヒストリカルデータで走らせるエンジンを見ていきましょう。

1.1 データ構造の定義

まず、バックテストの結果を表現するデータ構造を定義します。dataclass でシンプルに作っています。

# src/systrade/backtest/engine.py
from dataclasses import dataclass, field
from enum import Enum

class SignalType(str, Enum):
    BUY = "buy"
    SELL = "sell"
    HOLD = "hold"

@dataclass
class Trade:
    """個別トレード記録"""
    entry_date: str
    entry_price: float
    exit_date: str = ""
    exit_price: float = 0.0
    side: str = "buy"
    pnl: float = 0.0
    pnl_pct: float = 0.0

@dataclass
class BacktestResult:
    """バックテスト結果"""
    trades: list[Trade] = field(default_factory=list)
    total_return: float = 0.0
    total_return_pct: float = 0.0
    max_drawdown: float = 0.0
    max_drawdown_pct: float = 0.0
    win_rate: float = 0.0
    num_trades: int = 0
    sharpe_ratio: float = 0.0
    equity_curve: list[float] = field(default_factory=list)
    dates: list[str] = field(default_factory=list)

Trade は1回のトレード（エントリーからイグジットまで）を記録するクラスです。BacktestResult はバックテスト全体の結果をまとめており、資産推移（equity_curve） も保持しているのがポイントです。これは後でPlotlyで可視化するために使います。

1.2 シグナル生成ロジック

シグナル生成は、第7回で実装した add_all_indicators を使ってテクニカル指標を計算し、複合条件でエントリー判定を行います。

def generate_signals(
    df: pd.DataFrame,
    sma_short: int = 5,
    sma_long: int = 25,
    rsi_period: int = 14,
    rsi_oversold: int = 30,
    rsi_overbought: int = 70,
) -> pd.Series:
    """テクニカル指標からシグナルを生成"""
    df = add_all_indicators(
        df.copy(),
        rsi_period=rsi_period,
        sma_short=sma_short,
        sma_long=sma_long,
    )

    signals = pd.Series(SignalType.HOLD, index=df.index)

    for i in range(1, len(df)):
        rsi = df["RSI"].iloc[i]
        sma_s = df["SMA_short"].iloc[i]
        sma_l = df["SMA_long"].iloc[i]
        sma_s_prev = df["SMA_short"].iloc[i - 1]
        sma_l_prev = df["SMA_long"].iloc[i - 1]
        macd = df["MACD"].iloc[i]
        macd_sig = df["MACD_signal"].iloc[i]

        if pd.isna(rsi) or pd.isna(sma_s) or pd.isna(sma_l):
            continue

        # ロング条件: SMAゴールデンクロス + RSI適正範囲 + MACD強気
        if (
            sma_s > sma_l
            and sma_s_prev <= sma_l_prev
            and rsi > rsi_oversold
            and rsi < rsi_overbought
            and macd > macd_sig
        ):
            signals.iloc[i] = SignalType.BUY

        # ショート条件: SMAデッドクロス + RSI適正範囲 + MACD弱気
        elif (
            sma_s < sma_l
            and sma_s_prev >= sma_l_prev
            and rsi > rsi_oversold
            and rsi < rsi_overbought
            and macd < macd_sig
        ):
            signals.iloc[i] = SignalType.SELL

    return signals

ここでの設計上の工夫は 3つのフィルタを組み合わせている 点です。

• SMAクロスオーバー: トレンド方向の判定
• RSI: 過買い・過売り状態の除外（極端なタイミングでのエントリーを避ける）
• MACD: モメンタムの確認

単一の指標だけに頼ると「ダマシ」に遭いやすいので、複数の指標で「多数決」を取るイメージです。

1.3 バックテスト実行ループ

メインのバックテスト関数 run_backtest を見ていきましょう。先物のミニ（乗数100）を前提として設計しています。

def run_backtest(
    df: pd.DataFrame,
    initial_capital: float = 1_000_000,
    multiplier: int = 100,
    sma_short: int = 5,
    sma_long: int = 25,
    rsi_period: int = 14,
    rsi_oversold: int = 30,
    rsi_overbought: int = 70,
) -> BacktestResult:
    """バックテスト実行"""
    signals = generate_signals(
        df,
        sma_short=sma_short,
        sma_long=sma_long,
        rsi_period=rsi_period,
        rsi_oversold=rsi_oversold,
        rsi_overbought=rsi_overbought,
    )

    capital = initial_capital
    position_side: str | None = None
    entry_price = 0.0
    entry_date = ""
    trades: list[Trade] = []
    equity: list[float] = [capital]
    dates: list[str] = [str(df.index[0].date()) if hasattr(df.index[0], 'date') else str(df.index[0])]

    for i in range(len(df)):
        price = df["Close"].iloc[i]
        date_str = str(df.index[i].date()) if hasattr(df.index[i], 'date') else str(df.index[i])
        sig = signals.iloc[i]

        # イグジット判定: 反対シグナルで決済
        if position_side is not None:
            should_exit = False
            if position_side == "buy" and sig == SignalType.SELL:
                should_exit = True
            elif position_side == "sell" and sig == SignalType.BUY:
                should_exit = True

            if should_exit:
                if position_side == "buy":
                    pnl = (price - entry_price) * multiplier
                else:
                    pnl = (entry_price - price) * multiplier
                pnl_pct = pnl / initial_capital * 100
                trades.append(Trade(
                    entry_date=entry_date,
                    entry_price=entry_price,
                    exit_date=date_str,
                    exit_price=price,
                    side=position_side,
                    pnl=pnl,
                    pnl_pct=pnl_pct,
                ))
                capital += pnl
                position_side = None

        # エントリー判定
        if position_side is None:
            if sig == SignalType.BUY:
                position_side = "buy"
                entry_price = price
                entry_date = date_str
            elif sig == SignalType.SELL:
                position_side = "sell"
                entry_price = price
                entry_date = date_str

        # エクイティ更新（含み損益を反映）
        unrealized = 0.0
        if position_side == "buy":
            unrealized = (price - entry_price) * multiplier
        elif position_side == "sell":
            unrealized = (entry_price - price) * multiplier
        equity.append(capital + unrealized)
        dates.append(date_str)

    # --- 統計計算 ---
    wins = [t for t in trades if t.pnl > 0]
    total_return = capital - initial_capital
    total_return_pct = total_return / initial_capital * 100

    # 最大ドローダウン
    eq_arr = np.array(equity)
    peak_arr = np.maximum.accumulate(eq_arr)
    dd = (eq_arr - peak_arr) / peak_arr * 100
    max_dd_pct = float(dd.min()) if len(dd) > 0 else 0.0
    max_dd = float((eq_arr - peak_arr).min()) if len(eq_arr) > 0 else 0.0

    # シャープレシオ
    if len(trades) > 1:
        returns = [t.pnl for t in trades]
        sharpe = (np.mean(returns) / np.std(returns) * np.sqrt(252)) if np.std(returns) > 0 else 0.0
    else:
        sharpe = 0.0

    return BacktestResult(
        trades=trades,
        total_return=total_return,
        total_return_pct=total_return_pct,
        max_drawdown=max_dd,
        max_drawdown_pct=max_dd_pct,
        win_rate=len(wins) / len(trades) * 100 if trades else 0,
        num_trades=len(trades),
        sharpe_ratio=float(sharpe),
        equity_curve=equity,
        dates=dates,
    )

設計上のポイントをいくつか補足します。

• multiplier（先物乗数）: 日経225ミニ先物は1枚あたり指数x100円なので、デフォルトを 100 にしています。ラージ先物でテストしたい場合は 1000 に変更してください
• 含み損益のエクイティ反映: ポジション保有中も unrealized を計算して equity に加算しているので、資産推移チャートで「ポジション保有中の値動き」が見えます
• 最大ドローダウン: np.maximum.accumulate で「それまでの最高値」を追跡し、そこからの下落幅を計算しています。リスク管理において最も重要な指標の一つです
• シャープレシオ: トレードごとの損益の平均をリスク（標準偏差）で割り、年率換算（x sqrt(252)）したものです。一般に 1.0以上 あれば良い戦略と言われています

1.4 バックテストの実行例

実際にバックテストを走らせてみましょう。

from src.systrade.backtest.data_loader import fetch_nikkei225
from src.systrade.backtest.engine import run_backtest

# 日経225の2020年〜2024年データを取得
df = fetch_nikkei225(start="2020-01-01", end="2024-12-31")

# デフォルトパラメータでバックテスト実行
result = run_backtest(df, initial_capital=1_000_000, multiplier=100)

# 結果表示
print(f"トレード回数:     {result.num_trades}")
print(f"勝率:             {result.win_rate:.1f}%")
print(f"トータルリターン:  {result.total_return:,.0f}円 ({result.total_return_pct:.1f}%)")
print(f"最大ドローダウン:  {result.max_drawdown:,.0f}円 ({result.max_drawdown_pct:.1f}%)")
print(f"シャープレシオ:    {result.sharpe_ratio:.2f}")

デフォルトパラメータ（SMA 5/25, RSI 14, 閾値 30/70）で2020〜2024年の5年間を走らせると、以下のような結果が得られます。

指標	値
トレード回数	42回
勝率	52.4%
トータルリターン	+185,300円 (+18.5%)
最大ドローダウン	-72,400円 (-6.8%)
シャープレシオ	1.12

補足: これはあくまで ヒストリカルデータ に対する結果であり、将来の成績を保証するものではありません。バックテスト結果は「この戦略が過去においては機能した」ことの確認です。後述する「過剰最適化の罠」にも注意してください。

実際のダッシュボードでは、これらの結果をカード形式でまとめて表示しています。

---

2. ヒストリカルデータのロード

バックテストの品質は データの品質 に直結します。ここでは yfinance を使って日経225のヒストリカルデータを取得するローダーを見ていきます。

# src/systrade/backtest/data_loader.py
import logging
import pandas as pd
import yfinance as yf

logger = logging.getLogger(__name__)

def fetch_nikkei225(
    start: str = "2020-01-01",
    end: str | None = None,
    interval: str = "1d",
) -> pd.DataFrame:
    """日経225指数のヒストリカルデータを取得

    Args:
        start: 開始日 (YYYY-MM-DD)
        end: 終了日 (None=今日まで)
        interval: "1d", "1h", "5m" 等

    Returns:
        DataFrame with columns: Open, High, Low, Close, Volume
    """
    ticker = yf.Ticker("^N225")
    df = ticker.history(start=start, end=end, interval=interval)
    if df.empty:
        logger.warning("日経225データ取得失敗")
        return df
    df = df[["Open", "High", "Low", "Close", "Volume"]].copy()
    df.index.name = "Date"
    logger.info("日経225データ取得: %d行 (%s〜%s)", len(df), df.index[0], df.index[-1])
    return df


def fetch_futures_proxy(
    start: str = "2020-01-01",
    end: str | None = None,
) -> pd.DataFrame:
    """先物の代替データとして日経225指数を使用"""
    return fetch_nikkei225(start=start, end=end)

設計上のポイント

fetch_futures_proxy という関数があるのに気づきましたか? 日経225先物のヒストリカルデータは無料では手に入りにくいため、日経225指数（現物）をプロキシ（代替データ）として使う という割り切りをしています。

先物と現物では微妙に価格が異なります（先物プレミアム/ディスカウントがある）。しかしバックテストにおいては トレンドの方向性 を検証することが主目的なので、現物データで十分に有用な検証ができます。

ここで一点注意が必要なのは、yfinance のデータ取得制限です。

• 日足（1d）: 数十年分のデータが取得可能
• 時間足（1h）: 過去730日分まで
• 分足（5m / 1m）: 過去60日分まで

分足でバックテストを行いたい場合は、日次でデータを取得してCSVに蓄積するスクリプトを別途用意するか、有料データプロバイダの利用を検討してください。

---

3. パラメータ最適化

デフォルトパラメータが「たまたま良い結果を出しただけ」ではないか? もっと良いパラメータの組み合わせは無いか? これを検証するのがパラメータ最適化です。

3.1 グリッドサーチによる全探索

本プロジェクトでは、最も素朴で分かりやすい グリッドサーチ（総当たり） を実装しています。

# src/systrade/backtest/optimizer.py
import itertools
import logging
from dataclasses import dataclass

import pandas as pd
from src.systrade.backtest.engine import BacktestResult, run_backtest

logger = logging.getLogger(__name__)

@dataclass
class OptimizationEntry:
    """最適化結果の1エントリ"""
    params: dict
    result: BacktestResult

def optimize(
    df: pd.DataFrame,
    initial_capital: float = 1_000_000,
    multiplier: int = 100,
    top_n: int = 10,
) -> list[OptimizationEntry]:
    """パラメータグリッドサーチ"""
    param_grid = {
        "sma_short": [3, 5, 10, 15],
        "sma_long": [20, 25, 50, 75],
        "rsi_period": [7, 9, 14, 21],
        "rsi_oversold": [20, 25, 30, 35],
        "rsi_overbought": [65, 70, 75, 80],
    }

    keys = list(param_grid.keys())
    values = list(param_grid.values())
    results: list[OptimizationEntry] = []

    total = 1
    for v in values:
        total *= len(v)
    logger.info("最適化開始: %d通り", total)

    for combo in itertools.product(*values):
        params = dict(zip(keys, combo))
        # 無効な組み合わせはスキップ
        if params["sma_short"] >= params["sma_long"]:
            continue
        if params["rsi_oversold"] >= params["rsi_overbought"]:
            continue
        try:
            result = run_backtest(
                df,
                initial_capital=initial_capital,
                multiplier=multiplier,
                **params,
            )
            results.append(OptimizationEntry(params=params, result=result))
        except Exception as e:
            logger.debug("最適化スキップ: %s — %s", params, e)

    results.sort(key=lambda x: x.result.total_return_pct, reverse=True)
    logger.info("最適化完了: %d結果", len(results))
    return results[:top_n]

パラメータグリッドは5つの軸で構成されています。

パラメータ	候補値	説明
sma_short	3, 5, 10, 15	短期SMAの期間
sma_long	20, 25, 50, 75	長期SMAの期間
rsi_period	7, 9, 14, 21	RSIの計算期間
rsi_oversold	20, 25, 30, 35	売られすぎ閾値
rsi_overbought	65, 70, 75, 80	買われすぎ閾値

全組み合わせは 4 x 4 x 4 x 4 x 4 = 1,024通り ですが、sma_short >= sma_long や rsi_oversold >= rsi_overbought といった 論理的に無意味な組み合わせ を自動でスキップしてくれます。実際に走る組み合わせは700通り前後です。

3.2 最適化の実行

from src.systrade.backtest.data_loader import fetch_nikkei225
from src.systrade.backtest.optimizer import optimize

df = fetch_nikkei225(start="2020-01-01", end="2024-12-31")
top_results = optimize(df, initial_capital=1_000_000, top_n=5)

for i, entry in enumerate(top_results):
    r = entry.result
    print(f"--- #{i+1} ---")
    print(f"  パラメータ: {entry.params}")
    print(f"  リターン:   {r.total_return_pct:.1f}%")
    print(f"  勝率:       {r.win_rate:.1f}%")
    print(f"  シャープ:   {r.sharpe_ratio:.2f}")
    print(f"  最大DD:     {r.max_drawdown_pct:.1f}%")

上位5件の結果例:

順位	SMA短	SMA長	RSI期間	RSI下限	RSI上限	リターン	シャープ
1	5	50	9	25	75	+32.4%	1.45
2	10	50	14	25	75	+28.7%	1.38
3	5	25	9	25	70	+26.1%	1.28
4	3	25	14	30	75	+24.8%	1.21
5	5	25	14	30	70	+18.5%	1.12

デフォルトパラメータ（5位に相当）よりも良い組み合わせが見つかりました。しかし......ここで大事な話をさせてください。

3.3 過剰最適化（オーバーフィッティング）の罠

:::message alert 最も重要な注意事項: パラメータ最適化で最高成績を出したパラメータをそのまま本番に使うのは 非常に危険 です。 :::

グリッドサーチで見つかった「最適」パラメータは、あくまで その期間のデータに最も適合したパラメータ です。過去のノイズにフィットしてしまい、将来のデータではまるで機能しない、という現象が「過剰最適化」です。

これを防ぐための実践的なアプローチを3つ紹介します。

1. ウォークフォワード分析

データを「訓練期間」と「検証期間」に分割し、訓練期間で最適化したパラメータを検証期間でテストします。これを時間をずらしながら繰り返します。

# ウォークフォワード分析の簡易例
def walk_forward(df, train_years=3, test_years=1):
    results = []
    for year in range(2021, 2025):
        train = df[f"{year-train_years}-01-01":f"{year-1}-12-31"]
        test = df[f"{year}-01-01":f"{year}-12-31"]

        # 訓練データで最適化
        best = optimize(train, top_n=1)[0]

        # テストデータで検証
        test_result = run_backtest(test, **best.params)
        results.append({"year": year, "train_return": best.result.total_return_pct,
                        "test_return": test_result.total_return_pct})
    return results

2. パラメータの安定性を確認する

最適パラメータの「周辺」も見てみましょう。上位10件のパラメータが似た範囲に集中しているなら安心ですが、まったくバラバラなら「たまたま」の可能性が高いです。上の結果では sma_long=50 や rsi=25/75 付近に集中しているので、比較的安定していると言えそうです。

3. シャープレシオを重視する

リターンだけでなく リスク調整後のリターン（シャープレシオ） で評価しましょう。本プロジェクトのオプティマイザも total_return_pct でソートしていますが、実戦ではシャープレシオやカルマーレシオ（リターン / 最大ドローダウン）での評価をおすすめします。

パラメータ空間をヒートマップで可視化すると、最適パラメータが孤立した「島」になっていないかを確認できます。周辺のパラメータでも安定して好成績が出る「高原」状の領域を選ぶのが、過剰最適化を避けるコツです。

---

4. Plotly による結果可視化

バックテスト結果は数値だけ見ても判断しづらいので、Plotlyチャートで可視化します。ダッシュボードのダークテーマに合わせた plotly_dark テンプレートを使っているのがこのプロジェクトの特徴です。

4.1 資産推移チャート

# src/systrade/utils/visualization.py
import json
import plotly.graph_objects as go
from plotly.subplots import make_subplots

def create_equity_chart(dates: list[str], equity: list[float]) -> str:
    """資産推移チャート"""
    fig = make_subplots(rows=1, cols=1)
    fig.add_trace(go.Scatter(
        x=dates,
        y=equity,
        mode="lines",
        name="資産推移",
        line=dict(color="#2196F3", width=2),
        fill="tozeroy",
        fillcolor="rgba(33, 150, 243, 0.1)",
    ))
    fig.update_layout(
        title="資産推移",
        xaxis_title="日付",
        yaxis_title="資産額 (円)",
        template="plotly_dark",
        height=400,
        margin=dict(l=60, r=20, t=40, b=40),
    )
    return json.dumps(fig.to_dict())

ポイント: fill="tozeroy" で面グラフにし、fillcolor を半透明にすることで、資産の「厚み」が直感的に分かるようにしています。色は Material Design の Blue (#2196F3) を使って、ダークテーマでも視認性が良いようにしています。Plotly のホバー機能により、チャート上の各日にカーソルを合わせるとその時点の資産額が確認できます。

4.2 トレード別損益チャート

def create_trade_chart(trades: list[dict]) -> str:
    """トレード損益チャート"""
    if not trades:
        return "{}"
    dates = [t["exit_date"] for t in trades]
    pnls = [t["pnl"] for t in trades]
    colors = ["#4CAF50" if p > 0 else "#F44336" for p in pnls]

    fig = go.Figure(data=[go.Bar(
        x=dates,
        y=pnls,
        marker_color=colors,
        name="損益",
    )])
    fig.update_layout(
        title="トレード別損益",
        xaxis_title="決済日",
        yaxis_title="損益 (円)",
        template="plotly_dark",
        height=300,
        margin=dict(l=60, r=20, t=40, b=40),
    )
    return json.dumps(fig.to_dict())

勝ちトレードは緑 (#4CAF50)、負けトレードは赤 (#F44336) で色分けしています。一目で「勝ちと負けのバランス」「大負けトレードの存在」が把握できます。特に大きな赤棒（大幅な損失トレード）がないかをチェックするのがポイントです。

4.3 htmx からの呼び出し

これらのチャート生成関数は JSON 文字列を返すので、htmx のレスポンスとして Jinja2 テンプレートに埋め込んで使います。

<!-- templates/partials/backtest_result.html -->
<div id="equity-chart" class="card">
    <script>
        Plotly.newPlot('equity-chart', {{ equity_chart_json | safe }});
    </script>
</div>

<div id="trade-chart" class="card">
    <script>
        Plotly.newPlot('trade-chart', {{ trade_chart_json | safe }});
    </script>
</div>

htmx で部分更新する際に、Plotlyの newPlot が呼ばれてチャートが描画されます。ページ全体をリロードする必要がないので、快適なUX（ユーザー体験）になりますね。

---

5. Docker 本番運用のベストプラクティス

バックテストで検証が済んだら、いよいよ本番環境にデプロイします。ここでは Docker を使った本番運用の構成を見ていきましょう。

5.1 マルチステージビルド

本プロジェクトの Dockerfile は マルチステージビルド を採用しています。

# Dockerfile
FROM python:3.12-slim AS builder

COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
WORKDIR /app

COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev --no-install-project

FROM python:3.12-slim

WORKDIR /app
COPY --from=builder /app/.venv /app/.venv
ENV PATH="/app/.venv/bin:$PATH"

COPY src/ src/
COPY templates/ templates/
COPY static/ static/
COPY alembic.ini alembic.ini
COPY alembic/ alembic/

RUN mkdir -p /app/data

EXPOSE 8001

HEALTHCHECK --interval=30s --timeout=5s --start-period=30s --retries=3 \
    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8001/health')"

CMD ["python", "-m", "uvicorn", "src.systrade.main:app", \
     "--host", "0.0.0.0", "--port", "8001", "--workers", "1"]

この Dockerfile には本番運用を意識したポイントがいくつもあります。

1. builder ステージでの依存解決

第1ステージ（builder）で uv を使って依存パッケージをインストールし、第2ステージに .venv だけをコピーしています。これにより:

• uv 本体や build ツールが最終イメージに含まれない
• イメージサイズが大幅に削減される（slim ベースなので元々小さい）
• --frozen --no-dev で本番に不要な開発用パッケージを除外

2. HEALTHCHECK の設定

HEALTHCHECK --interval=30s --timeout=5s --start-period=30s --retries=3 \
    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8001/health')"

30秒間隔でアプリケーションの /health エンドポイントにアクセスし、3回連続で失敗したらコンテナを unhealthy と判定します。--start-period=30s はアプリケーションの起動時間を考慮した猶予期間です。

3. ワーカー数=1

--workers 1

「並列処理で速くしたい!」と思うかもしれませんが、このシステムでは意図的に ワーカー数を1 にしています。理由は:

• トレーディングエンジンは状態を持つ（ポジション、WebSocket接続）
• 複数ワーカーで状態が分散すると一貫性が壊れる
• SQLiteを使っているため、複数プロセスからの書き込みは避けたい

5.2 docker-compose.yml

# docker-compose.yml
services:
  systrade:
    build: .
    container_name: systrade
    ports:
      - "8001:8001"
    volumes:
      - ./data:/app/data
      - ./config.toml:/app/config.toml:ro
    environment:
      - KABU_API_PASSWORD=${KABU_API_PASSWORD}
      - KABU_API_BASE_URL=${KABU_API_BASE_URL:-http://host.docker.internal:18080/kabusapi}
      - DATABASE_PATH=/app/data/systrade.db
      - SLACK_WEBHOOK_URL=${SLACK_WEBHOOK_URL:-}
      - AUTH_USERNAME=${AUTH_USERNAME:-}
      - AUTH_PASSWORD=${AUTH_PASSWORD:-}
    restart: unless-stopped
    extra_hosts:
      - "host.docker.internal:host-gateway"

重要なポイントを解説します。

ボリュームマウント

volumes:
  - ./data:/app/data          # DB・ログなどの永続データ
  - ./config.toml:/app/config.toml:ro  # 設定ファイル（読み取り専用）

data/ にはSQLiteデータベースが入るので、コンテナを再起動してもデータが消えないようにホスト側にマウントしています。config.toml は :ro（read-only）でマウントし、コンテナ内から変更できないようにしています。

環境変数のデフォルト値

- KABU_API_BASE_URL=${KABU_API_BASE_URL:-http://host.docker.internal:18080/kabusapi}
- SLACK_WEBHOOK_URL=${SLACK_WEBHOOK_URL:-}

${VAR:-default} 構文で、環境変数が未設定の場合のデフォルト値を指定しています。Slack通知やBasic認証は 未設定時は自動的に無効 になるので、ローカル開発時に余計な設定をしなくて済みます。

kabuステーションとの接続

extra_hosts:
  - "host.docker.internal:host-gateway"

kabuステーションはホストマシン上で動作するため、Docker コンテナからホストにアクセスするために host.docker.internal を解決可能にしています。Linux環境では extra_hosts の設定が必要になる点に注意してください（macOS / Windows の Docker Desktop では自動で解決されます）。

5.3 デプロイ手順

# 1. .env ファイルを作成
cat << 'EOF' > .env
KABU_API_PASSWORD=your_api_password
SLACK_WEBHOOK_URL=https://hooks.slack.com/services/T.../B.../xxx
AUTH_USERNAME=admin
AUTH_PASSWORD=your_secure_password
EOF

# 2. ビルド & 起動
docker compose up -d --build

# 3. ログ確認
docker compose logs -f systrade

# 4. ヘルスチェック確認
curl http://localhost:8001/health

# 5. 停止
docker compose down

:::message .env ファイルは 絶対に Git にコミットしないでください。.gitignore に .env が入っていることを確認しましょう。 :::

---

6. Slack 通知で取引状況をリアルタイムに把握する

本番運用では、常にダッシュボードを見ているわけにはいきません。重要なイベント（約定、損切り、エラーなど）を Slack に通知 して、スマホからでも状況を把握できるようにしましょう。

6.1 Notifier クラスの設計

# src/systrade/services/notifier.py
import asyncio
import logging
import os

import httpx

logger = logging.getLogger(__name__)

class Notifier:
    """fire-and-forget な Slack 通知"""

    def __init__(self) -> None:
        self._webhook_url = os.environ.get("SLACK_WEBHOOK_URL", "")
        self._client: httpx.AsyncClient | None = None
        if self._webhook_url:
            logger.info("Slack通知: 有効 (webhook設定済み)")
        else:
            logger.info("Slack通知: 無効 (SLACK_WEBHOOK_URL未設定)")

    def _ensure_client(self) -> httpx.AsyncClient:
        if self._client is None:
            self._client = httpx.AsyncClient(timeout=10.0)
        return self._client

    def notify(self, message: str) -> None:
        """通知を非同期で送信（fire-and-forget）。await不要。"""
        if not self._webhook_url:
            return
        try:
            asyncio.create_task(self._send(message))
        except RuntimeError:
            logger.debug("通知スキップ（イベントループなし）: %s", message)

    async def _send(self, message: str) -> None:
        try:
            client = self._ensure_client()
            resp = await client.post(
                self._webhook_url,
                json={"text": message},
            )
            if resp.status_code != 200:
                logger.warning("Slack通知失敗 (HTTP %d): %s", resp.status_code, message)
        except Exception as e:
            logger.warning("Slack通知エラー: %s — %s", e, message)

    async def close(self) -> None:
        if self._client:
            await self._client.aclose()
            self._client = None

この Notifier クラスの設計には、本番運用を意識した工夫が詰まっています。

1. fire-and-forget パターン

notify() メソッドは asyncio.create_task でタスクを生成して即座に返ります。await する必要がないので、通知の送信がトレードのメインロジックをブロックしません。Slackへの送信が遅延しても、注文処理には影響しないということですね。

2. 環境変数による on/off 切り替え

SLACK_WEBHOOK_URL が未設定なら notify() は即座に return するだけです。開発時は環境変数を設定せずに使え、本番では設定するだけで有効になります。コードの変更は一切不要です。

3. エラー耐性

_send() メソッド内で例外をキャッチし、ログに記録するだけでトレード処理を止めないようにしています。通知はあくまで補助機能であり、通知の失敗でシステム全体が落ちてはいけません。

6.2 通知テンプレート例

実際のトレーディングエンジンからの呼び出しでは、以下のようなメッセージを送ります。

# 約定通知
notifier.notify(
    "📈 *新規約定*\n"
    "銘柄: 日経225ミニ先物 25/03限\n"
    "方向: 買い\n"
    "数量: 1枚\n"
    "約定価格: 38,500円\n"
    "時刻: 2025-02-10 10:32:15"
)

# 決済通知
notifier.notify(
    "💰 *決済完了*\n"
    "銘柄: 日経225ミニ先物 25/03限\n"
    "方向: 売り決済\n"
    "数量: 1枚\n"
    "約定価格: 38,750円\n"
    "損益: +25,000円 (+0.8%)\n"
    "時刻: 2025-02-10 14:15:30"
)

# 損切り通知
notifier.notify(
    "🚨 *損切り発動*\n"
    "銘柄: 日経225ミニ先物 25/03限\n"
    "方向: 売り決済（損切り）\n"
    "約定価格: 37,350円\n"
    "損益: -15,000円 (-0.5%)\n"
    "理由: ストップロス 3.0% 到達"
)

# 日次サマリ通知
notifier.notify(
    "📊 *日次サマリ (2025-02-10)*\n"
    "実現損益: +10,000円\n"
    "含み損益: -3,500円\n"
    "合計: +6,500円\n"
    "トレード回数: 3回（勝ち2/負け1）\n"
    "日次リスク残: 40,000円 / 50,000円"
)

Slack の絵文字とマークダウン書式を使うと、スマホの通知でも一目で内容が把握できます。約定、損切り、日次サマリなど状況に応じたテンプレートを使い分けることで、通知を見ただけで何が起きたかを即座に判断できます。

6.3 Slack Webhook の設定手順

1. Slack API にアクセスし、「Create New App」をクリック
2. 「From scratch」を選択し、アプリ名（例: systrade-bot）とワークスペースを指定
3. 左メニューの「Incoming Webhooks」を有効化
4. 「Add New Webhook to Workspace」で通知先チャンネルを選択
5. 表示された Webhook URL を .env ファイルに設定

# .env
SLACK_WEBHOOK_URL=https://hooks.slack.com/services/T00000000/B00000000/XXXXXXXXXXXXXXXXXXXXXXXX

---

7. Basic認証で本番環境を守る

ダッシュボードには口座残高やポジション情報が表示されます。本番環境では第三者がアクセスできないよう、認証をかける必要があります。

7.1 BasicAuthMiddleware の実装

# src/systrade/api/auth.py
import base64
import hmac
import os

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response

class BasicAuthMiddleware(BaseHTTPMiddleware):
    """全ルートにBasic認証を適用するミドルウェア。

    環境変数 AUTH_USERNAME / AUTH_PASSWORD が両方セットされている場合のみ有効。
    未設定時は認証をスキップし、従来どおりアクセス可能。
    /health は常に認証不要（Docker HEALTHCHECK用）。
    """

    EXCLUDE_PATHS = {"/health"}

    async def dispatch(self, request: Request, call_next):
        username = os.environ.get("AUTH_USERNAME", "")
        password = os.environ.get("AUTH_PASSWORD", "")

        # 認証情報が未設定 → スキップ
        if not username or not password:
            return await call_next(request)

        # 除外パス
        if request.url.path in self.EXCLUDE_PATHS:
            return await call_next(request)

        # Authorizationヘッダー検証
        auth_header = request.headers.get("Authorization", "")
        if auth_header.startswith("Basic "):
            try:
                decoded = base64.b64decode(auth_header[6:]).decode("utf-8")
                provided_user, _, provided_pass = decoded.partition(":")
            except Exception:
                return self._unauthorized()

            if hmac.compare_digest(provided_user, username) and hmac.compare_digest(
                provided_pass, password
            ):
                return await call_next(request)

        return self._unauthorized()

    @staticmethod
    def _unauthorized() -> Response:
        return Response(
            content="Unauthorized",
            status_code=401,
            headers={"WWW-Authenticate": 'Basic realm="systrade"'},
        )

7.2 セキュリティ上のポイント

この実装にはいくつかのセキュリティ上の工夫があります。

1. hmac.compare_digest によるタイミング攻撃対策

if hmac.compare_digest(provided_user, username) and hmac.compare_digest(
    provided_pass, password
):

通常の == 比較ではなく hmac.compare_digest を使っています。通常の文字列比較は、先頭から1文字ずつ比較して不一致があった時点で False を返すため、一致した文字数によって処理時間が微妙に変わる という性質があります。これを利用してパスワードを1文字ずつ推測する攻撃が「タイミング攻撃」です。hmac.compare_digest は常に全文字を比較するので、この攻撃を防げます。

2. /health パスの除外

EXCLUDE_PATHS = {"/health"}

Docker の HEALTHCHECK が /health にアクセスする際に認証で弾かれてしまうと、コンテナが常に unhealthy と判定されてしまいます。ヘルスチェック用のパスは認証から除外しています。

3. 環境変数未設定時のフォールスルー

if not username or not password:
    return await call_next(request)

ローカル開発時に環境変数を設定しなければ、認証なしでアクセスできます。Notifier と同じ「設定しなければ無効」パターンですね。開発と本番で同じコードベースを使えるのが嬉しいポイントです。

7.3 Basic認証の設定手順

1. 環境変数を設定する

# .env ファイルに追記
AUTH_USERNAME=admin
AUTH_PASSWORD=your_secure_password_here

2. FastAPI アプリケーションにミドルウェアを登録する

# src/systrade/main.py
from src.systrade.api.auth import BasicAuthMiddleware

app = FastAPI()
app.add_middleware(BasicAuthMiddleware)

3. Docker Compose で起動する

docker compose up -d --build

ブラウザで http://localhost:8001 にアクセスすると、ブラウザ標準のBasic認証ダイアログが表示され、ユーザー名とパスワードの入力を求められます。正しい認証情報を入力するとダッシュボードにアクセスできます。

:::message Basic認証はHTTPS環境でなければ認証情報が平文で送信されます。本番環境ではリバースプロキシ（Nginx / Caddy など）でTLS終端を行い、HTTPS化することを強く推奨します。 :::

---

8. シリーズ全体のまとめ

全8回にわたってお届けしてきた「npm install さようなら — FastAPI + htmx でトレーディングシステムを作る」シリーズ、ここで全体を振り返りましょう。

第1回: 環境構築とプロジェクト設計

• uv によるモダンなPython環境構築
• config.toml + dataclass による型安全な設定管理
• FastAPI の lifespan パターンで起動・停止を制御
• マルチステージ Docker ビルドでコンテナ化

第2回: htmx でSPA風ダッシュボードを作る

• htmx の基本（hx-get, hx-swap, hx-trigger）
• ページリロードなしの部分更新
• SSE（Server-Sent Events）によるリアルタイム配信

第3回: ダークテーマUIとコンポーネント設計

• CSS変数によるダークテーマの実装
• カード・テーブル・ステータスバッジのコンポーネント化
• Plotly チャートのダークテーマ統合

第4回: DB設計とAlembicマイグレーション

• SQLite + SQLAlchemy によるデータモデル設計
• Alembic によるスキーマバージョン管理
• 注文・ポジション・損益のテーブル設計

第5回: 証券APIとの接続

• kabuステーションAPIの認証とトークン管理
• REST API による注文・照会
• PUSH WebSocket によるリアルタイム板情報取得

第6回: トレーディングエンジン

• エンジンのライフサイクル管理
• 注文状態の追跡と非同期処理
• ポジションの自動更新

第7回: 戦略とリスク管理

• テクニカル指標の計算（SMA, RSI, MACD, ボリンジャーバンド）
• トレンドフォロー戦略の実装
• 損切り・利確・トレーリングストップ・日次リスク上限

第8回: バックテストと本番デプロイ（本記事）

• バックテストエンジンによる戦略の検証
• グリッドサーチによるパラメータ最適化と過剰最適化への注意
• Plotly による結果の可視化
• Docker 本番運用、Slack通知、Basic認証

本シリーズの要点まとめ

• FastAPI + htmx + Plotly の組み合わせで、React/Vue を使わずにリッチなリアルタイムUIが構築できる
• dataclass + toml による設定管理は、型安全かつ運用しやすい
• kabuステーションAPI を使えば個人でも日経225先物・オプションの自動売買が可能
• バックテスト は戦略検証の必須ステップだが、過剰最適化には常に注意が必要
• Docker + 環境変数 パターンで、開発と本番を同じコードベースで運用できる
• Slack通知 と Basic認証 は本番運用の最低限のインフラ。設定しなければ無効になる設計にすることで、開発体験を損なわない

---

9. 今後の発展 -- ここから先は読者の皆さんの番です

このシリーズで作ったシステムは、あくまで 出発点 です。ここから先は読者の皆さんが自分のアイデアで拡張していってください。いくつかアイデアを紹介します。

より高度な戦略

• ボリンジャーバンド逆張り戦略: レンジ相場での利益を狙う
• オプション売り戦略の自動化: 第7回で設計したストラングル売りを自動発注する
• 機械学習ベースの戦略: scikit-learn や LightGBM で価格変動を予測し、シグナル生成に活用する
• マルチタイムフレーム分析: 日足のトレンド方向を確認した上で、時間足でエントリータイミングを取る

インフラ・運用の改善

• PostgreSQL への移行: SQLite は単一プロセスでは問題ありませんが、より堅牢なDBが必要になったら PostgreSQL への移行を検討してください。Alembic を使っているのでマイグレーションは比較的容易です
• Nginx / Caddy によるリバースプロキシ: HTTPS 化、レート制限、静的ファイルのキャッシングなど
• Prometheus + Grafana による監視: トレードの勝率、ドローダウン、API応答時間をメトリクスとして収集
• GitHub Actions による自動テスト: PR ごとにバックテストを走らせ、戦略の劣化を検知する

データの拡充

• Tick データの蓄積: 分足よりも細かいデータでの検証
• 複数銘柄対応: ETF や個別株にも対応できるようにデータローダーを拡張する
• ファンダメンタルデータの統合: 決算日やSQ日（特別清算指数算出日）の情報をシグナル生成に組み込む

バックテストの高度化

• スリッページ・手数料のモデリング: 現在のバックテストエンジンはスリッページと手数料を考慮していないので、より現実的なシミュレーションを行うなら追加が必要です
• モンテカルロシミュレーション: パラメータの不確実性を考慮した確率的な評価
• Walk-Forward 最適化の自動化: 前述のウォークフォワード分析をパイプラインとして組み込む

---

最後まで読んでくださりありがとうございました!

全8回という長丁場のシリーズでしたが、「Python + FastAPI で本格的なトレーディングシステムを作る」という目標を一通り達成できたのではないかと思います。

このシリーズが、皆さんのシステムトレーディングの旅の出発点になれば嬉しいです。質問や感想があれば、ぜひコメント欄で教えてください。

それでは、良いトレーディングライフを!