※ Bybit V5は機能追加のペースが速いAPIです。パラメータの最終的な正解は必ず公式ドキュメントで確認してください。本記事のコードはすべてテストネットで動作確認しています。
最初のbotは1ファイル200行で動きます。問題はそこから機能を足していったときに起きます。
| 症状 | 原因 | 設計での対処 |
|---|---|---|
| バックテストと本番で挙動が違う | 戦略ロジックを2回書いている | Strategyインターフェースの共有 |
| WebSocketのコールバックが詰まる | 受信処理の中で発注している | イベントバスで受信と処理を分離 |
| リスク管理が戦略に散らばる | 各戦略が個別に損切りを判断 | RiskManagerに集約 |
| 停止しても注文が残る | 終了処理が無い / 例外で飛ぶ | graceful shutdown |
| テストが書けない | 全部がグローバル変数と副作用 | 依存性注入とモック |
| 1つの戦略のバグで全部止まる | 例外が伝播する | コンポーネント単位の例外隔離 |
バックテスト用のコードと本番用のコードを別々に書くと、必ず食い違います。指標の計算式、シグナルの判定条件、ポジションサイズの決め方——どこか1箇所でも違えば、バックテストの結果は本番の予測になりません。
しかも食い違いに気づけません。「バックテストでは勝つのに本番で負ける」の原因がコストでもルックアヘッドでもなく単純な実装の差だった、というのは実際によくあります。
だからこの記事の中心は「1つのStrategyクラスを両方で動かす」ことです。記事31で Strategy 基底クラスを作ったのは、このためでした。
┌──────────────┐ MarketData ┌──────────────────┐
│ DataFeed │ ────────────────> │ StrategyEngine │
│ (WS/REST) │ BarClosed │ (記事31のStrategy)│
└──────────────┘ └──────────────────┘
│ │ Signal
│ Fill / PositionUpdate v
│ ┌──────────────────┐
│ │ RiskManager │
│ │ (記事23/24) │
│ └──────────────────┘
│ │ OrderRequest
v v
┌──────────────┐ ┌──────────────────┐
│PositionKeeper│ <──────────────── │ OrderRouter │
│ (記事20) │ 確認 │ (記事13/17) │
└──────────────┘ └──────────────────┘
│ │
└────────────> Persistence <─────────┘
(記事17の注文台帳)
すべてのやり取りは EventBus 経由。
コンポーネントは互いを直接呼ばない（疎結合）。| コンポーネント | 責務 | してはいけないこと |
|---|---|---|
| DataFeed | WebSocket/RESTからイベントを生成 | 発注・戦略判断 |
| StrategyEngine | 確定足を受けてシグナルを出す | 直接APIを呼ぶ・リスク判断 |
| RiskManager | シグナルをサイズ付き注文に変換、上限を強制 | シグナルの生成 |
| OrderRouter | 検証・丸め・冪等化・レート制御して発注 | 戦略判断 |
| PositionKeeper | ポジション・残高の現在状態を保持 | 発注 |
| Persistence | 注文台帳・イベントログの永続化 | 判断 |
Strategyが http.place_order() を呼べる設計にすると、バックテストで動かせなくなります（ネットワークが必要になるため）。Strategyは「シグナルを返すだけ」にしてください。
この制約があるからこそ、同じStrategyクラスをバックテストエンジンと本番Runnerの両方に差し込めます。制約は不便ではなく、テスト可能性を保証する仕組みです。
"""core/events.py — コンポーネント間で受け渡すイベントの定義
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations
import time
from dataclasses import dataclass, field
from decimal import Decimal
from enum import Enum, auto
import pandas as pd
class EventType(Enum):
TICK = auto() # 価格更新
BAR_CLOSED = auto() # 確定足
ORDER_BOOK = auto() # 板更新
SIGNAL = auto() # 戦略のシグナル
ORDER_REQUEST = auto() # 発注要求（リスク検査後）
ORDER_UPDATE = auto() # 注文状態の変化
FILL = auto() # 約定
POSITION_UPDATE = auto() # ポジション変化
WALLET_UPDATE = auto() # 残高変化
RISK_STATE = auto() # リスク状態の変化
HEARTBEAT = auto() # 生存確認
SHUTDOWN = auto() # 停止要求
@dataclass(frozen=True)
class Event:
type: EventType
ts: float = field(default_factory=time.time)
source: str = ""
@property
def age_ms(self) -> float:
return (time.time() - self.ts) * 1000
@dataclass(frozen=True)
class TickEvent(Event):
symbol: str = ""
last: Decimal = Decimal(0)
bid: Decimal = Decimal(0)
ask: Decimal = Decimal(0)
mark: Decimal = Decimal(0)
@property
def mid(self) -> Decimal:
if self.bid and self.ask:
return (self.bid + self.ask) / 2
return self.last
@dataclass(frozen=True)
class BarClosedEvent(Event):
"""確定足。StrategyEngineが受け取る主要なイベント。"""
symbol: str = ""
interval: str = ""
bar_start: pd.Timestamp | None = None
open: float = 0.0
high: float = 0.0
low: float = 0.0
close: float = 0.0
volume: float = 0.0
class SignalAction(str, Enum):
ENTER_LONG = "enter_long"
ENTER_SHORT = "enter_short"
EXIT = "exit"
FLATTEN = "flatten"
ADJUST = "adjust"
@dataclass(frozen=True)
class SignalEvent(Event):
"""戦略が出すシグナル。数量ではなく「意図」を表す。
実際の数量はRiskManagerが決める（責務の分離）。
"""
symbol: str = ""
action: SignalAction = SignalAction.EXIT
strategy: str = ""
strength: float = 1.0 # 0〜1。サイズ調整のヒント
reason: str = ""
target_ratio: Decimal | None = None # 資金に対する目標比率
@dataclass(frozen=True)
class OrderRequestEvent(Event):
"""リスク検査を通った発注要求。OrderRouterが処理する。"""
symbol: str = ""
side: str = ""
qty: Decimal = Decimal(0)
order_type: str = "Market"
price: Decimal | None = None
reduce_only: bool = False
strategy: str = ""
purpose: str = "entry"
time_in_force: str | None = None
ttl_bars: int | None = None
@dataclass(frozen=True)
class FillEvent(Event):
symbol: str = ""
side: str = ""
qty: Decimal = Decimal(0)
price: Decimal = Decimal(0)
fee: Decimal = Decimal(0)
is_maker: bool = False
exec_pnl: Decimal = Decimal(0)
order_link_id: str = ""
strategy: str = ""
@dataclass(frozen=True)
class PositionUpdateEvent(Event):
symbol: str = ""
side: str = ""
size: Decimal = Decimal(0)
entry_price: Decimal = Decimal(0)
unrealised_pnl: Decimal = Decimal(0)
@dataclass(frozen=True)
class WalletUpdateEvent(Event):
total_equity: Decimal = Decimal(0)
available: Decimal = Decimal(0)
mm_rate: Decimal = Decimal(0)
class RiskLevel(Enum):
"""縮退の段階。記事24のサーキットブレーカーと対応する。"""
NORMAL = auto()
NO_NEW_ENTRY = auto()
REDUCE_ONLY = auto()
FLATTEN = auto()
HALTED = auto()
def allows_entry(self) -> bool:
return self is RiskLevel.NORMAL
def allows_exit(self) -> bool:
return self is not RiskLevel.HALTED
@dataclass(frozen=True)
class RiskStateEvent(Event):
level: RiskLevel = RiskLevel.NORMAL
reasons: tuple[str, ...] = ()
@dataclass(frozen=True)
class ShutdownEvent(Event):
reason: str = ""
flatten: bool = False # ポジションも決済するか"""core/bus.py — asyncioベースのイベントバス
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
設計上の要点:
- publish は絶対にブロックしない（WSコールバックから呼べる）
- 溢れたら古いものを捨てる（相場データは鮮度が命）
- 1つのハンドラの例外が他に伝播しない
"""
from __future__ import annotations
import asyncio
import logging
import time
from collections import defaultdict
from collections.abc import Awaitable, Callable
from .events import Event, EventType
log = logging.getLogger(__name__)
Handler = Callable[[Event], Awaitable[None] | None]
class EventBus:
def __init__(self, maxsize: int = 10_000, drop_oldest: bool = True):
self._queue: asyncio.Queue[Event] = asyncio.Queue(maxsize=maxsize)
self._handlers: dict[EventType, list[tuple[int, Handler]]] = \
defaultdict(list)
self._drop_oldest = drop_oldest
self._loop: asyncio.AbstractEventLoop | None = None
self._running = False
self.published = 0
self.processed = 0
self.dropped = 0
self.handler_errors = 0
self.max_lag_ms = 0.0
# ---- 購読 -----------------------------------------------------------
def subscribe(self, event_type: EventType, handler: Handler,
priority: int = 100) -> None:
"""priorityが小さいほど先に呼ばれる（リスク検査を先に通すため）。"""
self._handlers[event_type].append((priority, handler))
self._handlers[event_type].sort(key=lambda x: x[0])
# ---- 発行 -----------------------------------------------------------
def publish(self, event: Event) -> None:
"""同期版。WebSocketのコールバックから安全に呼べる。"""
self.published += 1
try:
self._queue.put_nowait(event)
except asyncio.QueueFull:
self.dropped += 1
if self._drop_oldest:
try:
self._queue.get_nowait()
self._queue.put_nowait(event)
except (asyncio.QueueEmpty, asyncio.QueueFull):
pass
def publish_threadsafe(self, event: Event) -> None:
"""別スレッド（pybitのWSコールバック等）から呼ぶ場合。"""
if self._loop is None:
self.publish(event)
return
self._loop.call_soon_threadsafe(self.publish, event)
async def apublish(self, event: Event) -> None:
"""バックプレッシャをかけたい場合（履歴の再生など）。"""
self.published += 1
await self._queue.put(event)
# ---- 実行 -----------------------------------------------------------
async def run(self) -> None:
self._loop = asyncio.get_running_loop()
self._running = True
log.info("EventBus started")
while self._running:
event = await self._queue.get()
lag = event.age_ms
self.max_lag_ms = max(self.max_lag_ms, lag)
for _, handler in self._handlers.get(event.type, []):
try:
result = handler(event)
if asyncio.iscoroutine(result):
await result
except asyncio.CancelledError:
raise
except Exception:
# 1つのハンドラの失敗で全体を止めない
self.handler_errors += 1
log.exception("handler failed for %s", event.type.name)
self.processed += 1
self._queue.task_done()
def stop(self) -> None:
self._running = False
# ---- 監視 -----------------------------------------------------------
@property
def backlog(self) -> int:
return self._queue.qsize()
def stats(self) -> dict:
return {
"published": self.published,
"processed": self.processed,
"dropped": self.dropped,
"backlog": self.backlog,
"handler_errors": self.handler_errors,
"max_lag_ms": round(self.max_lag_ms, 1),
}
def reset_max_lag(self) -> None:
self.max_lag_ms = 0.0subscribe(..., priority=10) のように優先度を指定できるようにしているのは、RiskManagerを他のハンドラより先に走らせるためです。
例えば WALLET_UPDATE で維持率が危険水準になったとき、RiskManagerが先に状態を REDUCE_ONLY に落としてから、StrategyEngineがシグナルを出す——という順序を保証できます。逆順だと、危険を認識する前に新規エントリーが飛びます。
"""core/component.py — コンポーネントの共通ライフサイクル
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
すべてのコンポーネントは start / stop / health を持つ。
Runnerはこの契約だけを知っていればよい（疎結合）。
"""
from __future__ import annotations
import abc
import logging
import time
from dataclasses import dataclass, field
from .bus import EventBus
log = logging.getLogger(__name__)
@dataclass
class Health:
name: str
ok: bool
detail: str = ""
last_activity: float = 0.0
@property
def idle_sec(self) -> float:
return time.time() - self.last_activity if self.last_activity else -1.0
def line(self) -> str:
mark = "OK " if self.ok else "NG "
idle = f"{self.idle_sec:.0f}s前" if self.last_activity else "-"
return f" [{mark}] {self.name:<18} 最終活動 {idle:>8} {self.detail}"
class Component(abc.ABC):
"""すべてのコンポーネントの基底クラス。"""
name: str = "component"
def __init__(self, bus: EventBus):
self.bus = bus
self._started = False
self.last_activity = 0.0
self.errors = 0
# ---- ライフサイクル ---------------------------------------------------
async def start(self) -> None:
if self._started:
return
self.register()
await self.on_start()
self._started = True
log.info("%s started", self.name)
async def stop(self) -> None:
if not self._started:
return
try:
await self.on_stop()
except Exception:
log.exception("%s の停止処理で例外", self.name)
self._started = False
log.info("%s stopped", self.name)
def register(self) -> None:
"""バスへの購読登録。サブクラスで実装する。"""
async def on_start(self) -> None:
"""起動処理（RESTでの初期化など）。"""
async def on_stop(self) -> None:
"""停止処理（接続の解放、状態の保存）。"""
# ---- 補助 -----------------------------------------------------------
def touch(self) -> None:
self.last_activity = time.time()
def health(self) -> Health:
return Health(name=self.name, ok=self._started and self.errors == 0,
detail=f"errors={self.errors}",
last_activity=self.last_activity)
def safe(self, fn, *args, **kwargs):
"""例外を飲み込んでカウントする（コンポーネント単位の隔離）。"""
try:
return fn(*args, **kwargs)
except Exception:
self.errors += 1
log.exception("%s でエラー", self.name)
return None"""components/feed.py — WebSocketの受信をイベントに変換する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
責務: 受信して publish するだけ。判断も発注もしない。
pybitのWSコールバックは別スレッドなので publish_threadsafe を使う。
"""
from __future__ import annotations
import logging
from decimal import Decimal
import pandas as pd
from core.bus import EventBus
from core.component import Component
from core.events import (
BarClosedEvent, EventType, FillEvent, PositionUpdateEvent, TickEvent,
WalletUpdateEvent,
)
log = logging.getLogger(__name__)
def _d(v, default: str = "0") -> Decimal:
if v in (None, "", "null"):
return Decimal(default)
return Decimal(str(v))
class MarketDataFeed(Component):
"""public WebSocketからticker/klineを受け、イベントに変換する。"""
name = "market_feed"
def __init__(self, bus: EventBus, client, symbols: list[str],
interval: str = "60", category: str = "linear",
warmup_bars: int = 300):
super().__init__(bus)
self.c = client
self.symbols = symbols
self.interval = interval
self.category = category
self.warmup_bars = warmup_bars
self.ws = None
self.history: dict[str, pd.DataFrame] = {}
self._last_bar_start: dict[str, int] = {}
async def on_start(self) -> None:
# 1) RESTで履歴をウォームアップ（WSは「今から先」しか来ない）
from kline_fetch import fetch_klines
for sym in self.symbols:
df = fetch_klines(self.c.http, sym, interval=self.interval,
category=self.category, verbose=False)
self.history[sym] = df.tail(self.warmup_bars)
log.info("%s: 履歴 %d本をウォームアップ", sym,
len(self.history[sym]))
# 2) WSを購読
from pybit.unified_trading import WebSocket
env = self.c.env.value
self.ws = WebSocket(
channel_type=self.category,
testnet=(env == "testnet"),
demo=False, # demoのpublicは本番を使う（記事04）
)
for sym in self.symbols:
self.ws.ticker_stream(symbol=sym, callback=self._on_ticker)
self.ws.kline_stream(interval=int(self.interval), symbol=sym,
callback=self._on_kline)
async def on_stop(self) -> None:
if self.ws:
try:
self.ws.exit()
except Exception:
pass
# ---- WSコールバック（別スレッドで呼ばれる） -------------------------
def _on_ticker(self, msg: dict) -> None:
self.touch()
d = msg.get("data") or {}
symbol = d.get("symbol")
if not symbol:
return
# linearのtickerは差分配信なので、無い項目は0で来る（記事26）
self.bus.publish_threadsafe(TickEvent(
type=EventType.TICK, source=self.name, symbol=symbol,
last=_d(d.get("lastPrice")), bid=_d(d.get("bid1Price")),
ask=_d(d.get("ask1Price")), mark=_d(d.get("markPrice")),
))
def _on_kline(self, msg: dict) -> None:
self.touch()
topic = msg.get("topic", "")
symbol = topic.split(".")[-1] if "." in topic else ""
for k in msg.get("data", []):
if not k.get("confirm"):
continue # 未確定足は無視（記事26）
start = int(k["start"])
if self._last_bar_start.get(symbol) == start:
continue # 重複配信を弾く
self._last_bar_start[symbol] = start
ts = pd.Timestamp(start, unit="ms", tz="UTC")
row = {
"open": float(k["open"]), "high": float(k["high"]),
"low": float(k["low"]), "close": float(k["close"]),
"volume": float(k["volume"]),
"turnover": float(k.get("turnover") or 0),
}
# 履歴に追記（Strategyはこの履歴を受け取る）
df = self.history.get(symbol)
if df is not None:
df.loc[ts] = row
self.history[symbol] = df.tail(self.warmup_bars * 2)
self.bus.publish_threadsafe(BarClosedEvent(
type=EventType.BAR_CLOSED, source=self.name,
symbol=symbol, interval=self.interval, bar_start=ts,
**row,
))
class AccountFeed(Component):
"""private WebSocketから約定・ポジション・残高を受ける（記事27）。"""
name = "account_feed"
def __init__(self, bus: EventBus, client):
super().__init__(bus)
self.c = client
self.ws = None
self._seen_exec: set[str] = set()
async def on_start(self) -> None:
from pybit.unified_trading import WebSocket
env = self.c.env.value
self.ws = WebSocket(
channel_type="private",
testnet=(env == "testnet"), demo=(env == "demo"),
api_key=self.c.cred.api_key, api_secret=self.c.cred.api_secret,
)
self.ws.execution_stream(callback=self._on_execution)
self.ws.position_stream(callback=self._on_position)
self.ws.wallet_stream(callback=self._on_wallet)
async def on_stop(self) -> None:
if self.ws:
try:
self.ws.exit()
except Exception:
pass
def _on_execution(self, msg: dict) -> None:
self.touch()
for e in msg.get("data", []):
if e.get("execType") != "Trade":
continue
exec_id = e.get("execId", "")
if exec_id and exec_id in self._seen_exec:
continue
if exec_id:
self._seen_exec.add(exec_id)
if len(self._seen_exec) > 20_000:
self._seen_exec.clear()
link_id = e.get("orderLinkId", "")
self.bus.publish_threadsafe(FillEvent(
type=EventType.FILL, source=self.name,
symbol=e.get("symbol", ""), side=e.get("side", ""),
qty=_d(e.get("execQty")), price=_d(e.get("execPrice")),
fee=_d(e.get("execFee")), is_maker=bool(e.get("isMaker")),
exec_pnl=_d(e.get("execPnl")), order_link_id=link_id,
strategy=link_id.split("-")[0] if "-" in link_id else "",
))
def _on_position(self, msg: dict) -> None:
self.touch()
for p in msg.get("data", []):
self.bus.publish_threadsafe(PositionUpdateEvent(
type=EventType.POSITION_UPDATE, source=self.name,
symbol=p.get("symbol", ""), side=p.get("side", ""),
size=_d(p.get("size")),
# WSは entryPrice（RESTは avgPrice）。記事20参照
entry_price=_d(p.get("entryPrice") or p.get("avgPrice")),
unrealised_pnl=_d(p.get("unrealisedPnl")),
))
def _on_wallet(self, msg: dict) -> None:
self.touch()
for w in msg.get("data", []):
self.bus.publish_threadsafe(WalletUpdateEvent(
type=EventType.WALLET_UPDATE, source=self.name,
total_equity=_d(w.get("totalEquity")),
available=_d(w.get("totalAvailableBalance")),
mm_rate=_d(w.get("accountMMRate")),
))ここがこの記事の中心です。記事31で定義した Strategy 基底クラスを、本番でもそのまま動かします。
"""components/strategy_engine.py — Strategyを本番で動かすアダプタ
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
要点: 記事31の Strategy.on_bar(history, position, equity) を
そのまま呼ぶ。Strategy側は本番かバックテストかを知らない。
バックテスト: エンジンが履歴をスライスして渡す
本番: DataFeedが積んだ履歴を渡す
-> 同じコードが両方で動く
"""
from __future__ import annotations
import logging
from decimal import Decimal
import pandas as pd
from bt.types import Order, OrderType, Position, Side # 記事31の型
from core.bus import EventBus
from core.component import Component
from core.events import (
BarClosedEvent, Event, EventType, FillEvent, OrderRequestEvent,
RiskLevel, RiskStateEvent,
)
log = logging.getLogger(__name__)
class StrategyEngine(Component):
"""確定足を受けてStrategyを呼び、注文要求をバスに流す。"""
name = "strategy_engine"
def __init__(self, bus: EventBus, strategies: dict[str, object],
feed, position_keeper, min_bars: int = 100):
"""strategies: {symbol: Strategyインスタンス}"""
super().__init__(bus)
self.strategies = strategies
self.feed = feed
self.keeper = position_keeper
self.min_bars = min_bars
self.risk_level = RiskLevel.NORMAL
self.bars_processed = 0
self.signals_emitted = 0
def register(self) -> None:
self.bus.subscribe(EventType.BAR_CLOSED, self.on_bar_closed,
priority=200)
self.bus.subscribe(EventType.FILL, self.on_fill, priority=200)
self.bus.subscribe(EventType.RISK_STATE, self.on_risk_state,
priority=10)
async def on_start(self) -> None:
for symbol, strat in self.strategies.items():
history = self.feed.history.get(symbol, pd.DataFrame())
self.safe(strat.on_start, history)
# ---- イベントハンドラ -------------------------------------------------
def on_risk_state(self, event: Event) -> None:
assert isinstance(event, RiskStateEvent)
if event.level is not self.risk_level:
log.warning("リスク状態: %s -> %s (%s)",
self.risk_level.name, event.level.name,
" / ".join(event.reasons))
self.risk_level = event.level
def on_fill(self, event: Event) -> None:
assert isinstance(event, FillEvent)
strat = self.strategies.get(event.symbol)
if strat is None:
return
# 記事31のStrategyと同じFill型に変換して渡す
from bt.types import Fill
fill = Fill(
ts=pd.Timestamp(event.ts, unit="s", tz="UTC"),
side=Side.BUY if event.side == "Buy" else Side.SELL,
qty=float(event.qty), price=float(event.price),
fee=float(event.fee), is_maker=event.is_maker,
tag=event.strategy,
)
self.safe(strat.on_fill, fill)
def on_bar_closed(self, event: Event) -> None:
assert isinstance(event, BarClosedEvent)
self.touch()
self.bars_processed += 1
strat = self.strategies.get(event.symbol)
if strat is None:
return
history = self.feed.history.get(event.symbol)
if history is None or len(history) < self.min_bars:
return # ウォームアップ中
# 現在のポジションと純資産をStrategyに渡す
pos = self._position_for(event.symbol)
equity = float(self.keeper.equity())
orders = self.safe(strat.on_bar, history, pos, equity)
if not orders:
return
for order in orders:
self._emit(event.symbol, order, type(strat).__name__)
# ---- 変換 -----------------------------------------------------------
def _position_for(self, symbol: str) -> Position:
"""記事31のPosition型に変換して渡す（Strategyは同じ型を見る）。"""
p = self.keeper.get(symbol)
if p is None:
return Position()
return Position(qty=float(p.signed_size),
entry_price=float(p.avg_price))
def _emit(self, symbol: str, order: Order, strategy: str) -> None:
"""StrategyのOrderを、リスク検査に渡すOrderRequestに変換する。"""
# リスク状態によって新規エントリーを止める
is_entry = not order.reduce_only
if is_entry and not self.risk_level.allows_entry():
log.info("[%s] リスク状態 %s のため新規を抑止: %s",
strategy, self.risk_level.name, order.tag)
return
if not self.risk_level.allows_exit():
return
self.bus.publish(OrderRequestEvent(
type=EventType.ORDER_REQUEST, source=self.name,
symbol=symbol,
side="Buy" if order.side is Side.BUY else "Sell",
qty=Decimal(str(order.qty)),
order_type=("Market" if order.order_type is OrderType.MARKET
else "Limit"),
price=(Decimal(str(order.price))
if order.price is not None else None),
reduce_only=order.reduce_only,
strategy=strategy, purpose=order.tag or "signal",
time_in_force=("PostOnly"
if order.order_type is OrderType.POST_ONLY
else None),
ttl_bars=order.ttl_bars,
))
self.signals_emitted += 1
def health(self):
h = super().health()
h.detail = (f"bars={self.bars_processed} "
f"signals={self.signals_emitted} "
f"risk={self.risk_level.name}")
return hStrategyEngine がやっているのは型の変換だけです。Strategy側の on_bar(history, position, equity) というシグネチャは、バックテストエンジンと完全に同じです。
この設計の効果は具体的です。バックテストで検証したStrategyクラスを、1行も変更せずに本番で動かせます。逆に本番で見つけたバグを直せば、バックテストの結果も自動的に正しくなります。二重管理が消えます。
コンポーネントを組み立て、確実に停止するのがRunnerの役割です。停止処理を怠ると、注文とポジションが放置されます。
"""core/runner.py — コンポーネントの組み立てと graceful shutdown
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
停止の順序が重要:
1) 新規発注を止める（DataFeedとStrategyを先に止める）
2) 未約定注文をキャンセル
3) 必要ならポジションを決済
4) 状態を永続化
5) 接続を解放
逆順にすると、停止処理中に新しい注文が入る。
"""
from __future__ import annotations
import asyncio
import contextlib
import logging
import signal
import time
from .bus import EventBus
from .component import Component, Health
from .events import EventType, RiskLevel, RiskStateEvent, ShutdownEvent
log = logging.getLogger(__name__)
class BotRunner:
def __init__(self, bus: EventBus, heartbeat_sec: float = 30.0,
stale_threshold_sec: float = 120.0):
self.bus = bus
self.components: list[Component] = []
self.heartbeat_sec = heartbeat_sec
self.stale_threshold_sec = stale_threshold_sec
self._tasks: list[asyncio.Task] = []
self._shutdown = asyncio.Event()
self._shutdown_reason = ""
self._flatten_on_exit = False
self.started_at = 0.0
def add(self, *components: Component) -> "BotRunner":
"""追加順が起動順、逆順が停止順になる。"""
self.components.extend(components)
return self
# ---- シグナル -------------------------------------------------------
def _install_signal_handlers(self) -> None:
loop = asyncio.get_running_loop()
def request(sig_name: str) -> None:
log.warning("シグナル %s を受信。停止します", sig_name)
self.request_shutdown(f"signal:{sig_name}")
for sig in (signal.SIGINT, signal.SIGTERM):
try:
loop.add_signal_handler(sig, request, sig.name)
except NotImplementedError:
# Windowsでは add_signal_handler が使えない
signal.signal(sig, lambda s, f: request(signal.Signals(s).name))
def request_shutdown(self, reason: str, flatten: bool = False) -> None:
if not self._shutdown.is_set():
self._shutdown_reason = reason
self._flatten_on_exit = flatten
self.bus.publish(ShutdownEvent(type=EventType.SHUTDOWN,
source="runner", reason=reason,
flatten=flatten))
self._shutdown.set()
# ---- 監視 -----------------------------------------------------------
def health(self) -> list[Health]:
return [c.health() for c in self.components]
async def _heartbeat(self) -> None:
while not self._shutdown.is_set():
await asyncio.sleep(self.heartbeat_sec)
uptime = time.time() - self.started_at
stats = self.bus.stats()
log.info("稼働 %.0f分 / bus %s", uptime / 60, stats)
for h in self.health():
if not h.ok:
log.warning("不健全: %s", h.line())
# データが途絶えたら新規を止める（記事24の思想）
for c in self.components:
if c.last_activity and \
(time.time() - c.last_activity) > self.stale_threshold_sec:
log.error("%s が %.0f秒間無活動。新規を停止します",
c.name, time.time() - c.last_activity)
self.bus.publish(RiskStateEvent(
type=EventType.RISK_STATE, source="runner",
level=RiskLevel.NO_NEW_ENTRY,
reasons=(f"{c.name}のデータ停止",),
))
if stats["backlog"] > 5000:
log.error("イベントキューが滞留 (%d)。処理が追いついていません",
stats["backlog"])
self.bus.reset_max_lag()
# ---- 本体 -----------------------------------------------------------
async def run(self) -> int:
self.started_at = time.time()
self._install_signal_handlers()
# 起動（追加順）
for c in self.components:
try:
await c.start()
except Exception:
log.exception("%s の起動に失敗。停止します", c.name)
await self._shutdown_sequence()
return 1
self._tasks = [
asyncio.create_task(self.bus.run(), name="bus"),
asyncio.create_task(self._heartbeat(), name="heartbeat"),
]
log.info("=" * 56)
log.info("稼働開始: %d コンポーネント", len(self.components))
log.info("=" * 56)
try:
await self._shutdown.wait()
except asyncio.CancelledError:
self._shutdown_reason = "cancelled"
finally:
await self._shutdown_sequence()
return 0
async def _shutdown_sequence(self) -> None:
log.info("停止処理を開始します（理由: %s）",
self._shutdown_reason or "不明")
# 1) コンポーネントを逆順に停止（DataFeed/Strategyが先に止まる）
for c in reversed(self.components):
with contextlib.suppress(Exception):
await asyncio.wait_for(c.stop(), timeout=15.0)
# 2) バスに残ったイベントを処理しきる（最大5秒）
try:
await asyncio.wait_for(self.bus._queue.join(), timeout=5.0)
except (asyncio.TimeoutError, Exception):
log.warning("バスの残イベント %d 件を破棄します", self.bus.backlog)
self.bus.stop()
for t in self._tasks:
t.cancel()
for t in self._tasks:
with contextlib.suppress(asyncio.CancelledError, Exception):
await t
uptime = time.time() - self.started_at
log.info("停止完了（稼働 %.1f分 / %s）", uptime / 60,
self.bus.stats())
# ---- エントリポイント ---------------------------------------------------
async def main() -> int:
import logging.config
from components.feed import AccountFeed, MarketDataFeed
from components.order_router import OrderRouterComponent
from components.position import PositionKeeperComponent
from components.risk import RiskManagerComponent
from components.strategy_engine import StrategyEngine
from bt.strategy import SmaCrossStrategy
from client import make_client
from config_model import load_settings
from order_ledger import OrderLedger
from reconciler import startup_reconcile
from idempotent_router import IdempotentRouter
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)-7s %(name)-22s %(message)s",
)
settings = load_settings()
c = make_client(settings.env)
bus = EventBus(maxsize=settings.bus_maxsize)
# --- 起動前のリコンサイル（記事17）。これを飛ばしてはいけない -------
ledger = OrderLedger()
router = IdempotentRouter(c.http, ledger)
try:
startup_reconcile(c, ledger, router,
strategy_prefixes=set(settings.strategy_prefixes),
cancel_orphans=settings.cancel_orphans,
require_clean=True)
except RuntimeError as e:
log.error("リコンサイル失敗のため起動を中止します: %s", e)
return 1
# --- コンポーネントの組み立て -------------------------------------
feed = MarketDataFeed(bus, c, symbols=settings.symbols,
interval=settings.interval)
account = AccountFeed(bus, c)
keeper = PositionKeeperComponent(bus, c)
risk = RiskManagerComponent(bus, keeper, settings.risk)
order_router = OrderRouterComponent(bus, c, router, keeper)
strategies = {
sym: SmaCrossStrategy(**settings.strategy_params)
for sym in settings.symbols
}
engine = StrategyEngine(bus, strategies, feed, keeper)
runner = BotRunner(bus, heartbeat_sec=settings.heartbeat_sec)
# 起動順: 状態を持つものを先に、判断するものを後に
runner.add(keeper, account, feed, risk, order_router, engine)
return await runner.run()
if __name__ == "__main__":
import sys
sys.exit(asyncio.run(main()))_shutdown_sequence() がコンポーネントを逆順に止めているのが要点です。追加順が keeper → account → feed → risk → router → engine なので、停止は engine → router → risk → feed → account → keeper になります。
つまり最初に止まるのはStrategyEngine（新しいシグナルが出なくなる）、次にOrderRouter（未約定注文のキャンセル処理を on_stop で行う）、最後にPositionKeeper（状態の保存）です。この順序が逆だと、キャンセル処理の最中に新しい注文が飛びます。
"""config_model.py — TOMLベースの設定管理（環境別オーバーライド付き）
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
方針:
- 設定はコードに書かない（環境ごとに変えるため）
- 起動時に検証する（不正な設定で動き始めない）
- APIキーは.envから（設定ファイルには入れない）
"""
from __future__ import annotations
import os
import tomllib
from dataclasses import dataclass, field
from decimal import Decimal
from pathlib import Path
CONFIG_DIR = Path("config")
@dataclass
class RiskSettings:
max_notional_per_symbol: Decimal = Decimal("1000")
max_total_notional: Decimal = Decimal("5000")
max_daily_loss_pct: Decimal = Decimal("3.0")
max_consecutive_losses: int = 5
max_account_mm_rate: Decimal = Decimal("0.35")
risk_per_trade_pct: Decimal = Decimal("0.5")
max_orders_per_minute: int = 30
def validate(self) -> list[str]:
errs = []
if self.max_notional_per_symbol <= 0:
errs.append("max_notional_per_symbol は正の値が必要")
if self.max_total_notional < self.max_notional_per_symbol:
errs.append("max_total_notional < max_notional_per_symbol")
if not (0 < self.max_daily_loss_pct <= 100):
errs.append("max_daily_loss_pct は 0〜100")
if not (0 < self.max_account_mm_rate < 1):
errs.append("max_account_mm_rate は 0〜1")
return errs
@dataclass
class Settings:
env: str = "testnet"
symbols: list[str] = field(default_factory=lambda: ["BTCUSDT"])
category: str = "linear"
interval: str = "60"
leverage: Decimal = Decimal("3")
strategy_params: dict = field(default_factory=dict)
strategy_prefixes: list[str] = field(default_factory=lambda: ["bot"])
risk: RiskSettings = field(default_factory=RiskSettings)
bus_maxsize: int = 10_000
heartbeat_sec: float = 30.0
cancel_orphans: bool = False
log_level: str = "INFO"
def validate(self) -> None:
errs: list[str] = []
if self.env not in ("testnet", "demo", "mainnet"):
errs.append(f"env が不正: {self.env}")
if not self.symbols:
errs.append("symbols が空")
if self.category not in ("linear", "inverse", "spot"):
errs.append(f"category が不正: {self.category}")
if self.leverage <= 0:
errs.append("leverage は正の値が必要")
errs += self.risk.validate()
# 本番では上限を厳しく強制する
if self.env == "mainnet":
if self.risk.max_total_notional > Decimal("100000"):
errs.append("本番のmax_total_notionalが大きすぎます"
"（意図的なら設定ファイルを見直してください）")
if self.leverage > Decimal("10"):
errs.append("本番のleverageが10倍を超えています")
if errs:
raise ValueError("設定が不正です:\n - " + "\n - ".join(errs))
def banner(self) -> str:
live = " ***** 実資金 *****" if self.env == "mainnet" else ""
return "\n".join([
f"環境 : {self.env}{live}",
f"銘柄 : {', '.join(self.symbols)} ({self.category})",
f"時間足 : {self.interval}",
f"レバレッジ : {self.leverage}x",
f"1銘柄上限 : {self.risk.max_notional_per_symbol} USDT",
f"合計上限 : {self.risk.max_total_notional} USDT",
f"日次損失上限 : {self.risk.max_daily_loss_pct}%",
f"戦略パラメータ: {self.strategy_params}",
])
def _merge(base: dict, override: dict) -> dict:
out = dict(base)
for k, v in override.items():
if isinstance(v, dict) and isinstance(out.get(k), dict):
out[k] = _merge(out[k], v)
else:
out[k] = v
return out
def load_settings(env: str | None = None,
config_dir: Path = CONFIG_DIR) -> Settings:
"""base.toml を読み、{env}.toml で上書きする。"""
env = env or os.getenv("BYBIT_ENV", "testnet")
data: dict = {}
base = config_dir / "base.toml"
if base.exists():
data = tomllib.loads(base.read_text(encoding="utf-8"))
env_file = config_dir / f"{env}.toml"
if env_file.exists():
data = _merge(data, tomllib.loads(env_file.read_text(encoding="utf-8")))
risk_data = data.pop("risk", {})
risk = RiskSettings(**{
k: (Decimal(str(v)) if k not in ("max_consecutive_losses",
"max_orders_per_minute") else int(v))
for k, v in risk_data.items()
})
if "leverage" in data:
data["leverage"] = Decimal(str(data["leverage"]))
settings = Settings(env=env, risk=risk,
**{k: v for k, v in data.items()
if k in Settings.__dataclass_fields__
and k not in ("env", "risk")})
settings.validate()
return settings
SAMPLE_BASE_TOML = '''# config/base.toml — 全環境の共通設定
symbols = ["BTCUSDT", "ETHUSDT"]
category = "linear"
interval = "60"
leverage = 3
strategy_prefixes = ["trendfol"]
bus_maxsize = 10000
heartbeat_sec = 30
log_level = "INFO"
[strategy_params]
fast = 20
slow = 60
risk_pct = 0.5
[risk]
max_notional_per_symbol = 1000
max_total_notional = 5000
max_daily_loss_pct = 3.0
max_consecutive_losses = 5
max_account_mm_rate = 0.35
risk_per_trade_pct = 0.5
max_orders_per_minute = 30
'''
SAMPLE_MAINNET_TOML = '''# config/mainnet.toml — 本番だけ上限を厳しくする
symbols = ["BTCUSDT"]
leverage = 2
cancel_orphans = false
[risk]
max_notional_per_symbol = 300
max_total_notional = 600
max_daily_loss_pct = 1.5
max_orders_per_minute = 10
'''
if __name__ == "__main__":
CONFIG_DIR.mkdir(exist_ok=True)
for name, content in (("base.toml", SAMPLE_BASE_TOML),
("mainnet.toml", SAMPLE_MAINNET_TOML)):
p = CONFIG_DIR / name
if not p.exists():
p.write_text(content, encoding="utf-8")
print(f"生成: {p}")
for env in ("testnet", "mainnet"):
print(f"\n=== {env} ===")
try:
print(load_settings(env).banner())
except ValueError as e:
print(f"設定エラー: {e}")bybit-bot/
├── .env # APIキー（.gitignore）
├── .gitignore # .env / .venv / data / LIVE_ENABLED
├── LIVE_ENABLED # 本番許可トークン（VPSのみ・記事04）
├── requirements.txt
├── pyproject.toml
│
├── config/
│ ├── base.toml # 共通設定
│ ├── testnet.toml
│ ├── demo.toml
│ └── mainnet.toml # 本番だけ上限を厳しく
│
├── core/ # 基盤（取引所に依存しない）
│ ├── events.py # イベント型（サンプル1）
│ ├── bus.py # イベントバス（サンプル2）
│ ├── component.py # ライフサイクル（サンプル3）
│ └── runner.py # 組み立て・停止（サンプル6）
│
├── exchange/ # Bybit固有の層
│ ├── config.py # 環境・キー（記事01・04）
│ ├── signer.py # 署名（記事02）
│ ├── client.py # HTTPクライアント（記事01・02）
│ ├── rate_limit.py # トークンバケット（記事05）
│ ├── errors.py # 例外階層（記事06）
│ └── symbol_rules.py # 銘柄仕様・丸め（記事03）
│
├── components/ # 実行時のコンポーネント
│ ├── feed.py # DataFeed（サンプル4）
│ ├── strategy_engine.py # Strategyアダプタ（サンプル5）
│ ├── risk.py # RiskManager（記事23・24）
│ ├── order_router.py # 発注（記事13・17）
│ └── position.py # PositionKeeper（記事20・27）
│
├── strategies/ # 戦略（バックテストと共有）
│ ├── base.py # Strategy基底（記事31）
│ ├── sma_cross.py
│ ├── trend_follow.py # 記事33
│ └── grid.py # 記事36
│
├── bt/ # バックテスト（記事31・32）
│ ├── types.py
│ ├── engine.py
│ ├── fill.py
│ ├── report.py
│ └── validation.py # WFA / DSR / PBO（記事32）
│
├── data/ # 永続化（.gitignore）
│ ├── kline/*.parquet # 記事07
│ ├── orders.sqlite # 注文台帳（記事17）
│ └── history.sqlite # 約定履歴（記事18）
│
├── scripts/ # 単発の運用スクリプト
│ ├── diagnose.py # 自己診断（記事06）
│ ├── sync_history.py # 履歴同期（記事18）
│ ├── flatten_all.py # 緊急全決済（記事20）
│ └── tax_report.py # 年間損益（記事40）
│
└── tests/
├── conftest.py # フィクスチャ（モック取引所）
├── test_signer.py # 署名（記事02）
├── test_symbol_rules.py # 丸め（記事03）
├── test_idempotency.py # 冪等性（記事17）
├── test_bus.py # イベントバス
├── test_strategy.py # 戦略ロジック（ネットワーク不要）
└── test_lookahead.py # 未来参照の検出（記事31）"""tests/test_bus.py — イベントバスとコンポーネントのテスト
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
pytest tests/test_bus.py -v
"""
from __future__ import annotations
import asyncio
import pytest
from core.bus import EventBus
from core.component import Component
from core.events import Event, EventType
@pytest.fixture
def bus() -> EventBus:
return EventBus(maxsize=100)
@pytest.mark.asyncio
async def test_publish_and_handle(bus):
received: list[Event] = []
bus.subscribe(EventType.TICK, received.append)
task = asyncio.create_task(bus.run())
bus.publish(Event(type=EventType.TICK, source="test"))
await asyncio.sleep(0.05)
bus.stop()
task.cancel()
assert len(received) == 1
assert bus.processed == 1
@pytest.mark.asyncio
async def test_priority_order(bus):
"""priorityが小さいハンドラが先に呼ばれること。"""
order: list[str] = []
bus.subscribe(EventType.TICK, lambda e: order.append("late"), priority=200)
bus.subscribe(EventType.TICK, lambda e: order.append("early"), priority=10)
task = asyncio.create_task(bus.run())
bus.publish(Event(type=EventType.TICK))
await asyncio.sleep(0.05)
bus.stop()
task.cancel()
assert order == ["early", "late"]
@pytest.mark.asyncio
async def test_handler_error_does_not_stop_bus(bus):
"""1つのハンドラが例外を投げても、他のハンドラと後続処理が続くこと。"""
ok: list[int] = []
def broken(e: Event) -> None:
raise RuntimeError("intentional")
bus.subscribe(EventType.TICK, broken, priority=10)
bus.subscribe(EventType.TICK, lambda e: ok.append(1), priority=20)
task = asyncio.create_task(bus.run())
for _ in range(3):
bus.publish(Event(type=EventType.TICK))
await asyncio.sleep(0.1)
bus.stop()
task.cancel()
assert len(ok) == 3 # 例外の後も処理が続いた
assert bus.handler_errors == 3
@pytest.mark.asyncio
async def test_async_handler(bus):
received: list[Event] = []
async def handler(e: Event) -> None:
await asyncio.sleep(0)
received.append(e)
bus.subscribe(EventType.FILL, handler)
task = asyncio.create_task(bus.run())
bus.publish(Event(type=EventType.FILL))
await asyncio.sleep(0.05)
bus.stop()
task.cancel()
assert len(received) == 1
def test_drop_oldest_on_full():
"""溢れたら古いイベントを捨て、最新を保持すること。"""
small = EventBus(maxsize=3, drop_oldest=True)
for _ in range(10):
small.publish(Event(type=EventType.TICK))
assert small.published == 10
assert small.dropped == 7
assert small.backlog == 3 # 常に最新3件だけ残る
def test_publish_never_blocks():
"""publishがブロックしないこと（WSコールバックから呼べる）。"""
import time
small = EventBus(maxsize=1)
t0 = time.time()
for _ in range(10_000):
small.publish(Event(type=EventType.TICK))
assert time.time() - t0 < 1.0 # 1万件でも1秒未満
@pytest.mark.asyncio
async def test_component_lifecycle(bus):
"""start/stopが1回だけ実行されること。"""
class Dummy(Component):
name = "dummy"
def __init__(self, bus):
super().__init__(bus)
self.starts = 0
self.stops = 0
async def on_start(self) -> None:
self.starts += 1
async def on_stop(self) -> None:
self.stops += 1
d = Dummy(bus)
await d.start()
await d.start() # 2回目は無視される
await d.stop()
await d.stop()
assert d.starts == 1
assert d.stops == 1
@pytest.mark.asyncio
async def test_component_error_isolation(bus):
"""safe()が例外を飲み込んでカウントすること。"""
class Dummy(Component):
name = "dummy"
d = Dummy(bus)
def boom():
raise ValueError("x")
assert d.safe(boom) is None
assert d.errors == 1
assert d.health().ok is False| テストの層 | 対象 | ネットワーク | 実行頻度 |
|---|---|---|---|
| 単体 | 署名・丸め・イベントバス・戦略ロジック | 不要 | 毎コミット（CI） |
| モック結合 | 発注ラッパ・冪等化・リコンサイル | 不要（記事04のモック） | 毎コミット |
| リプレイ | 記録したWSデータでの戦略挙動 | 不要（記事04） | 日次 |
| testnet結合 | 実際のAPIレスポンス形式 | 必要 | 週次 / リリース前 |
| デモ運用 | 戦略の実挙動・運用の安定性 | 必要 | 常時（記事04） |
asyncio.Queue.put_nowait() を直接呼ぶのは安全ではありません。サンプル2で publish_threadsafe() を用意しているのはこのためで、loop.call_soon_threadsafe() 経由でイベントループに渡します。SignalEvent が strength や target_ratio を持つのは、Strategyがヒントだけを渡すためです。equity を受け取って数量を返す形になっていますが、RiskManagerが最終的にクリップするため上限は守られます。maxsize を大きく、または別バスにする）のが安全です。またそもそも溢れているのは処理能力不足のシグナルなので、dropped をメトリクスとして監視し、増えたら通知してください（記事39）。strategies = {symbol: Strategy} なので1銘柄1戦略です。同じ銘柄で複数戦略を動かすには、{(symbol, strategy_name): Strategy} に変える必要がありますが、ポジションが合算されるため各戦略の損益が分離できません（記事20のFAQ参照）。orderLinkId のプレフィックスで按分集計する（記事18）かの選択になります。前者のほうが明確です。request_shutdown(reason, flatten=True) のように、停止理由によって挙動を分けるのが正解です。投資は自己判断・自己責任でお願いいたします。本ブログの情報・ツール・ソフトウェアを参考・使用して、取引をされて何らかの損失を被られても、当サイトでは責任は一切負いません（間接的・付随的な障害・損失も含め一切の責任を負いません）。自己責任でご利用ください。
当サイトからダウンロードしたプログラム・EAは本人利用に限り完全無料でご利用いただけます。プログラムの全てまたは一部の複製・コピー・リバースエンジニアリング・改変・2次配布・商用利用・転売・販売等は一切禁止いたします。違反を発見した場合は法的処置を行います。
当サイト上に存在するソフトウェア・ツール・文章・画像・動画等の著作物を無断転載することを禁止します。引用の範囲を超えるものについては、法的処置を行います。
当サイトで配布するツール・ソフトウェアは無料提供ですのでサポートなどは行っておりません(逆に丁寧にサポートしているサイトはほとんど違法サイトですのでご注意ください。投資助言・代理業登録なしでサポート・助言することは法律で禁止されています)。万が一、ツールやソフトウェアに不具合や問題があっても当サイトでは修正・対応などができない場合があります。ご利用の環境でトラブルが発生しても当サイトでは助言することはできませんので自己責任でご利用ください。検索するなどして自分で解決してください。
当サイトでは取引所・証券会社へのリンクを貼っていますが、これは情報として提供するものであり、勧誘行為などは一切行っておりません。そのため、取引所・証券会社の登録方法・使い方・オススメ金融機関・オススメ金融商品などを説明した記事はサイト内にありませんので、ご自身で調べた上で判断ください(本当はサイト内で丁寧に説明したいのですが、仲介・取次ぎにあたる可能性があり第一種金融商品取引業登録を行っていない場合、違法となる可能性があるため出来ません。ご容赦ください)。
当サイトからリンクやバナーなどによって他のサイトに移動された場合、移動先サイトで提供される情報、サービス等について一切の責任を負いません。当サイト掲載情報やリンク先の掲載情報の合法性、正確性、道徳性、最新性、適切性、著作権の許諾や有無など、その内容については一切の保証を致しかねます。
当サイトは、ご訪問いただいた時間、訪問者のアクセス元のホスト名、IPアドレス、OSとブラウザのバージョン、リンク元のURL、モニターの情報等のパソコン環境情報を収集し、アクセス解析をしております。この目的は訪問者が検索サイトからご訪問いただいた場合、どのような検索で当サイトに来訪いただいたかの解析にのみ使用しており、第3者に公開はいたしません。また、個人を特定するものではありません。