イグジットロジック¶
監視¶
PositionMonitor¶
class PositionMonitor:
"""ポジション監視"""
async def monitor_loop(self):
"""監視ループ"""
while True:
for position in self.open_positions:
# 1. 価格更新
await self._update_price(position)
# 2. イグジット判定
exit_signal = self._check_exit_conditions(position)
# 3. イグジット実行
if exit_signal:
await self._execute_exit(position, exit_signal)
await asyncio.sleep(60) # 1分間隔
イグジット条件¶
4条件¶
graph TD
A[ポジション監視] --> B{1. 利確?}
B -->|Yes| E[決済: PROFIT_TARGET]
B -->|No| C{2. 損切り?}
C -->|Yes| E[決済: STOP_LOSS]
C -->|No| D{3. 時間経過?}
D -->|Yes| E[決済: TIME_EXIT]
D -->|No| F{4. GEX変化?}
F -->|Yes| E[決済: GEX_CHANGE]
F -->|No| G[継続保有]
条件詳細¶
| 条件 | Spear | Shield |
|---|---|---|
| 利確 | +80% | +60% |
| 損切り | -30% | -50% |
| 時間 | 10日 | 30日 |
| GEX変化 | レジーム反転 | レジーム反転 |
利確判定¶
def check_profit_target(position: Position) -> bool:
"""利確条件をチェック"""
# クレジットスプレッドの場合
# 価値が下がる = 利益
current_value = position.current_price
entry_value = position.entry_price
# 利益率計算
profit_pct = (entry_value - current_value) / entry_value * 100
return profit_pct >= position.strategy.profit_target
損切り判定¶
def check_stop_loss(position: Position) -> bool:
"""損切り条件をチェック"""
current_value = position.current_price
entry_value = position.entry_price
# 損失率計算
loss_pct = (current_value - entry_value) / entry_value * 100
return loss_pct >= position.strategy.stop_loss
時間イグジット¶
def check_time_exit(position: Position) -> bool:
"""時間イグジット条件をチェック"""
days_held = (date.today() - position.entry_date).days
return days_held >= position.strategy.max_hold_days
GEX変化イグジット¶
def check_gex_change(position: Position) -> bool:
"""GEX変化条件をチェック"""
current_regime = self.gex_engine.get_regime(position.symbol)
entry_regime = position.entry_regime
# レジーム反転をチェック
if entry_regime == "POS_GAMMA_SAFE" and current_regime == "NEG_GAMMA_RISKY":
return True
if entry_regime == "NEG_GAMMA_RISKY" and current_regime == "POS_GAMMA_SAFE":
return True
return False
決済実行¶
async def execute_exit(position: Position, reason: str):
"""決済を実行"""
# 1. 決済注文を作成
close_order = ComboOrder(
legs=[
OrderLeg(position.sell_option, "BUY", position.quantity),
OrderLeg(position.buy_option, "SELL", position.quantity),
],
order_type="MKT", # 市場注文
tif="DAY"
)
# 2. 注文発注
result = await ibkr.place_order(close_order)
# 3. トレード記録
trade_store.record_exit(
position_id=position.id,
exit_price=result.fill_price,
exit_reason=reason,
pnl=calculate_pnl(position, result.fill_price)
)
# 4. 通知
await notify_exit(position, reason)
PnL計算¶
def calculate_pnl(position: Position, exit_price: float) -> float:
"""PnLを計算"""
# クレジットスプレッドの場合
entry_credit = position.entry_price * 100 * position.quantity
exit_debit = exit_price * 100 * position.quantity
pnl = entry_credit - exit_debit
return pnl
イグジット追跡(v2.5)¶
FilterTracker統合¶
v2.5から、イグジット理由とPnLを自動追跡します。
from core.filter_tracker import FilterTracker, ExitReasonType
tracker = FilterTracker.instance()
# イグジット結果を記録
tracker.record_exit(
trade_id="abc123",
symbol="NVDA",
target_date=date.today(),
reason=ExitReasonType.PROFIT_TAKE.value,
pnl=450.0,
pnl_pct=30.0,
)
追跡されるイグジット理由¶
| 理由 | 説明 | 期待PnL |
|---|---|---|
| PROFIT_TAKE | 利確(目標達成) | +$$ |
| STOP_LOSS | 損切り(閾値到達) | -$$ |
| TIME_DECAY | 最大保有日数経過 | ±$ |
| EXPIRY | 満期接近(DTE閾値) | ±$ |
| GEX_CHANGE | GEXレジーム変化 | ±$$ |
| MOMENTUM_REVERSAL | モメンタム反転(Spear専用) | ±$$ |
イグジット統計¶
# サマリーからイグジット分布を取得
stats = tracker.get_summary()
# 出力例
{
"exit_distribution": [
{"name": "PROFIT_TAKE", "count": 10, "total_pnl": 4500.0, "avg_pnl": 450.0},
{"name": "STOP_LOSS", "count": 5, "total_pnl": -1500.0, "avg_pnl": -300.0},
{"name": "TIME_DECAY", "count": 3, "total_pnl": 200.0, "avg_pnl": 66.7},
...
]
}
取引分析エクスポート¶
特定の取引について詳細分析用データをエクスポートできます。
# 特定の取引をエクスポート
python scripts/export_trade_analysis.py --trade-ids abc123,def456
# 大勝ち/大負け取引を自動抽出
python scripts/export_trade_analysis.py --auto-select --threshold 500
実装ファイル¶
scripts/run_paper_trading.py- 監視ループcore/execution.py- 決済実行core/trade_store.py- トレード記録core/filter_tracker.py- イグジット追跡scripts/export_trade_analysis.py- 分析エクスポート