トレード状態機械（State Machine）

概要

Trade State Machine は、トレードのライフサイクルを明確な状態遷移として管理するモジュールです。

設計理念

「UW検知 → 即エントリー」ではなく、テクニカル条件を満たすまで「監視状態」で待機する。

---

状態遷移図

stateDiagram-v2
    [*] --> IDLE: 初期状態

    IDLE --> FLOW_DETECTED: UWフロー検知

    FLOW_DETECTED --> WATCHING_TECHNICAL: フロー検証OK
    FLOW_DETECTED --> IDLE: フロー検証NG

    WATCHING_TECHNICAL --> ENTRY_TRIGGERED: テクニカル条件成立
    WATCHING_TECHNICAL --> IDLE: タイムアウト(8時間)
    WATCHING_TECHNICAL --> WATCHING_TECHNICAL: 追加フロー検知(強化)

    ENTRY_TRIGGERED --> POSITION_OPEN: 約定成功
    ENTRY_TRIGGERED --> IDLE: 約定失敗

    POSITION_OPEN --> CLOSED: イグジット

    CLOSED --> [*]: 終了

---

状態定義

1. IDLE（待機）

初期状態: 何も監視していない

class TradeState(Enum):
    IDLE = "idle"
    FLOW_DETECTED = "flow_detected"
    WATCHING_TECHNICAL = "watching_technical"
    ENTRY_TRIGGERED = "entry_triggered"
    POSITION_OPEN = "position_open"
    CLOSED = "closed"

2. FLOW_DETECTED（フロー検知）

UWでフローを検知した直後の状態

• FlowValidatorでフロー品質を検証
• 検証OKならWATCHING_TECHNICALへ遷移
• 検証NGならIDLEへ戻る

3. WATCHING_TECHNICAL（テクニカル監視）

テクニカル条件の成立を待機中

• MTF分析で環境を確認
• 重要ラインのブレイクアウトを監視
• タイムアウト: 8時間（当日引けまで）

複数フローの強化

30分以内に同一銘柄で追加フローを検知した場合、信頼度スコアが上昇

4. ENTRY_TRIGGERED（エントリートリガー）

テクニカル条件が成立し、エントリー準備完了

• スプレッド構築
• 注文発注
• 約定確認

5. POSITION_OPEN（ポジション保有中）

ポジションを保有している状態

• 価格監視
• イグジット条件チェック
• ロール判定

6. CLOSED（クローズ済み）

トレード完了

• 結果記録
• 統計更新

---

実装詳細

core/trade_state_machine.py

@dataclass
class TradeContext:
    """トレードコンテキスト"""
    symbol: str
    state: TradeState
    initial_flow: OptionFlow
    additional_flows: List[OptionFlow] = field(default_factory=list)
    confidence_score: float = 0.0
    state_entered_at: datetime = field(default_factory=datetime.now)
    timeout_hours: float = 8.0  # 当日引けまで

class TradeStateMachine:
    """トレード状態機械"""

    def __init__(self):
        self._contexts: Dict[str, TradeContext] = {}

    def on_flow_detected(self, flow: OptionFlow) -> TradeContext:
        """フロー検知時の処理"""
        symbol = flow.ticker

        # 既存コンテキストがあれば強化
        if symbol in self._contexts:
            ctx = self._contexts[symbol]
            if ctx.state == TradeState.WATCHING_TECHNICAL:
                return self._reinforce_flow(ctx, flow)

        # 新規コンテキスト作成
        ctx = TradeContext(
            symbol=symbol,
            state=TradeState.FLOW_DETECTED,
            initial_flow=flow,
            confidence_score=self._calculate_initial_confidence(flow)
        )
        self._contexts[symbol] = ctx
        return ctx

    def on_flow_validated(self, symbol: str, is_valid: bool) -> TradeContext:
        """フロー検証結果の処理"""
        ctx = self._contexts.get(symbol)
        if not ctx:
            return None

        if is_valid:
            ctx.state = TradeState.WATCHING_TECHNICAL
            ctx.state_entered_at = datetime.now()
        else:
            ctx.state = TradeState.IDLE
            del self._contexts[symbol]

        return ctx

    def on_technical_trigger(self, symbol: str) -> TradeContext:
        """テクニカルトリガー成立時の処理"""
        ctx = self._contexts.get(symbol)
        if ctx and ctx.state == TradeState.WATCHING_TECHNICAL:
            ctx.state = TradeState.ENTRY_TRIGGERED
            return ctx
        return None

    def check_timeouts(self) -> List[str]:
        """タイムアウトチェック"""
        expired = []
        now = datetime.now()

        for symbol, ctx in list(self._contexts.items()):
            if ctx.state == TradeState.WATCHING_TECHNICAL:
                elapsed = (now - ctx.state_entered_at).total_seconds() / 3600
                if elapsed > ctx.timeout_hours:
                    ctx.state = TradeState.IDLE
                    del self._contexts[symbol]
                    expired.append(symbol)

        return expired

---

信頼度スコア

初期スコア計算

def _calculate_initial_confidence(self, flow: OptionFlow) -> float:
    """初期信頼度スコアを計算"""
    score = 0.0

    # プレミアム（大きいほど高スコア）
    if flow.premium >= 100000:
        score += 0.3
    elif flow.premium >= 50000:
        score += 0.2

    # Sweep判定
    if flow.is_sweep:
        score += 0.2

    # Ask Side
    if flow.side in ['ASK', 'ABOVE_ASK']:
        score += 0.2

    # OTM度合い（適度なOTMが高スコア）
    if 0.2 <= flow.delta <= 0.4:
        score += 0.15

    # 満期（適切なDTE）
    if 7 <= flow.dte <= 30:
        score += 0.15

    return min(score, 1.0)

フロー強化

def _reinforce_flow(self, ctx: TradeContext, new_flow: OptionFlow) -> TradeContext:
    """追加フローで信頼度を強化"""
    # 30分以内のフローのみ強化対象
    elapsed = (datetime.now() - ctx.state_entered_at).total_seconds() / 60
    if elapsed > 30:
        return ctx

    ctx.additional_flows.append(new_flow)

    # 同方向のフローで強化
    if new_flow.put_call == ctx.initial_flow.put_call:
        ctx.confidence_score = min(ctx.confidence_score + 0.15, 1.0)

    logger.info(f"[{ctx.symbol}] フロー強化: 信頼度 {ctx.confidence_score:.2f}")
    return ctx

---

タイムアウト設定

状態	タイムアウト	理由
FLOW_DETECTED	5分	検証は即座に完了すべき
WATCHING_TECHNICAL	8時間	当日引けまで
ENTRY_TRIGGERED	30秒	約定待ち

Gemini推奨

デイトレード〜短期スイングが前提なら、その日のうちにモメンタムが出なければ「鮮度が落ちた」とみなして監視リストから外す。

---

ペーパートレードでの使用

# run_paper_trading.py

class PaperTradingEngine:
    def __init__(self):
        self.state_machine = TradeStateMachine()
        self.flow_validator = FlowValidator()
        self.mtf_analyzer = MTFAnalyzer()

    async def on_uw_flow(self, flow: OptionFlow):
        # 1. 状態機械にフロー通知
        ctx = self.state_machine.on_flow_detected(flow)

        # 2. フロー検証
        validation = self.flow_validator.validate(flow, ...)
        ctx = self.state_machine.on_flow_validated(
            flow.ticker, 
            validation.is_valid
        )

        if ctx.state != TradeState.WATCHING_TECHNICAL:
            return

        # 3. テクニカル監視ループ開始
        asyncio.create_task(self._watch_technical(ctx))

    async def _watch_technical(self, ctx: TradeContext):
        while ctx.state == TradeState.WATCHING_TECHNICAL:
            # MTF分析
            mtf = self.mtf_analyzer.analyze(ctx.symbol, ...)

            if mtf.trigger_confirmed:
                self.state_machine.on_technical_trigger(ctx.symbol)
                await self._execute_entry(ctx)
                return

            # タイムアウトチェック
            if self.state_machine.check_timeouts():
                return

            await asyncio.sleep(60)  # 1分間隔

---

ログ出力例

[TSLA] 状態遷移: IDLE → FLOW_DETECTED
  フロー: CALL $250 12/20, プレミアム $125,000
  初期信頼度: 0.65

[TSLA] 状態遷移: FLOW_DETECTED → WATCHING_TECHNICAL
  フロー検証: VALIDATED

[TSLA] フロー強化: 追加フロー検知
  信頼度: 0.65 → 0.80

[TSLA] 状態遷移: WATCHING_TECHNICAL → ENTRY_TRIGGERED
  トリガー: 1時間足ブレイクアウト確認

[TSLA] 状態遷移: ENTRY_TRIGGERED → POSITION_OPEN
  約定: Put Credit Spread $245/$240

---

関連ファイル

ファイル	役割
core/trade_state_machine.py	状態機械本体
core/flow_validator.py	フロー検証
core/mtf_analyzer.py	MTF分析
core/technical_validator.py	テクニカル検証

---

更新履歴

日付	変更内容
2025-11-29	Gemini分析に基づき実装
2025-11-29	複数フロー強化ロジック追加
2025-11-29	タイムアウト設定（8時間）