"""Position monitoring loop (``docs/06-operational-flow.md`` §3). Walks every ``open`` position, builds the live snapshot, asks :func:`exit_decision.evaluate`, and — when the action is not ``HOLD`` — sends the inverse combo order to close. Decisions are persisted to the ``decisions`` table; transitions to the audit log. """ from __future__ import annotations import asyncio import json import logging from dataclasses import dataclass from datetime import UTC, datetime, timedelta from decimal import Decimal from typing import Any from uuid import uuid4 from cerbero_bite.clients.deribit import ComboLegOrder from cerbero_bite.core.exit_decision import ( PositionSnapshot, evaluate, ) from cerbero_bite.core.types import OptionLeg, PutOrCall from cerbero_bite.runtime.dependencies import RuntimeContext from cerbero_bite.state import ( DecisionRecord, DvolSnapshot, InstructionRecord, PositionRecord, transaction, ) from cerbero_bite.state import connect as connect_state __all__ = [ "MonitorCycleResult", "PositionOutcome", "run_monitor_cycle", ] _log = logging.getLogger("cerbero_bite.runtime.monitor") @dataclass(frozen=True) class PositionOutcome: proposal_id: str action: str closed: bool reason: str | None @dataclass(frozen=True) class MonitorCycleResult: outcomes: list[PositionOutcome] # --------------------------------------------------------------------------- # Snapshot helpers # --------------------------------------------------------------------------- async def _build_position_snapshot( ctx: RuntimeContext, *, record: PositionRecord, spot: Decimal, dvol: Decimal, return_4h: Decimal, eth_price_usd_now: Decimal, now: datetime, ) -> PositionSnapshot | None: """Fetch tickers for the two legs and build a :class:`PositionSnapshot`. Returns ``None`` if the broker no longer quotes one of the legs — the caller will arm an alert in that case. """ tickers = await ctx.deribit.get_tickers( [record.short_instrument, record.long_instrument] ) by_name: dict[str, dict[str, Any]] = { str(t.get("instrument_name")): t for t in tickers if isinstance(t, dict) } short = by_name.get(record.short_instrument) long_ = by_name.get(record.long_instrument) if short is None or long_ is None: return None short_mid = Decimal(str(short.get("mark_price") or 0)) long_mid = Decimal(str(long_.get("mark_price") or 0)) if short_mid <= 0 or long_mid <= 0: return None short_delta = Decimal( str((short.get("greeks") or {}).get("delta") or record.delta_at_entry) ) debit_per_contract = short_mid - long_mid debit_total = debit_per_contract * Decimal(record.n_contracts) legs = [ OptionLeg( instrument=record.short_instrument, side="SELL", strike=record.short_strike, expiry=record.expiry, type=_option_type_from_name(record.short_instrument), size=record.n_contracts, mid_price_eth=short_mid, delta=short_delta, gamma=Decimal(str((short.get("greeks") or {}).get("gamma") or 0)), theta=Decimal(str((short.get("greeks") or {}).get("theta") or 0)), vega=Decimal(str((short.get("greeks") or {}).get("vega") or 0)), ), OptionLeg( instrument=record.long_instrument, side="BUY", strike=record.long_strike, expiry=record.expiry, type=_option_type_from_name(record.long_instrument), size=record.n_contracts, mid_price_eth=long_mid, delta=Decimal(str((long_.get("greeks") or {}).get("delta") or 0)), gamma=Decimal(str((long_.get("greeks") or {}).get("gamma") or 0)), theta=Decimal(str((long_.get("greeks") or {}).get("theta") or 0)), vega=Decimal(str((long_.get("greeks") or {}).get("vega") or 0)), ), ] return PositionSnapshot( proposal_id=record.proposal_id, spread_type=record.spread_type, # type: ignore[arg-type] legs=legs, credit_received_eth=record.credit_eth, credit_received_usd=record.credit_usd, spot_at_entry=record.spot_at_entry, dvol_at_entry=record.dvol_at_entry, expiry=record.expiry, opened_at=record.opened_at or record.created_at, eth_price_usd_now=eth_price_usd_now, spot_now=spot, dvol_now=dvol, mark_combo_now_eth=debit_total, delta_short_now=short_delta, return_4h_now=return_4h, now=now, ) def _option_type_from_name(name: str) -> PutOrCall: suffix = name.rsplit("-", 1)[-1] if suffix not in ("P", "C"): raise ValueError(f"unknown option type suffix in {name!r}") return suffix # type: ignore[return-value] async def _fetch_return_4h(ctx: RuntimeContext, *, now: datetime) -> Decimal: """Compute ETH 4h return. Resolution order: 1. local ``dvol_history`` snapshot at least 3h30 old (recorded by previous monitor cycles); 2. Deribit ``get_historical`` 1h candles 4h ago — bootstrap when SQLite has no recent sample (first cycle after a fresh container, or after long downtime). Returns ``0`` only when both sources fail; in that case the monitor cycle emits a LOW alert and exit_decision falls back to HOLD on the adverse-move trigger. """ cutoff = now - timedelta(hours=3, minutes=30) floor = now - timedelta(hours=8) conn = connect_state(ctx.db_path) try: row = conn.execute( "SELECT timestamp, spot FROM dvol_history " "WHERE asset = 'ETH' AND timestamp <= ? AND timestamp >= ? " "ORDER BY timestamp DESC LIMIT 1", (cutoff.isoformat(), floor.isoformat()), ).fetchone() finally: conn.close() if row is not None: past_spot = Decimal(str(row[1])) if past_spot != 0: spot_now = await ctx.deribit.index_price_eth() return spot_now / past_spot - Decimal("1") # Fallback: ask Deribit for the 4h candle close. try: past_close = await ctx.deribit.historical_close( instrument="ETH-PERPETUAL", start=now - timedelta(hours=5), end=now - timedelta(hours=3, minutes=30), resolution="1h", ) except Exception: # pragma: no cover — defensive, surface as LOW alert past_close = None if past_close is None or past_close == 0: await ctx.alert_manager.low( source="monitor_cycle", message="no return_4h sample available (history empty + bootstrap failed)", ) return Decimal("0") spot_now = await ctx.deribit.index_price_eth() return spot_now / past_close - Decimal("1") # --------------------------------------------------------------------------- # Cycle entry point # --------------------------------------------------------------------------- async def run_monitor_cycle( ctx: RuntimeContext, *, now: datetime | None = None, ) -> MonitorCycleResult: """Walk every open position, evaluate exit, place close orders if needed.""" when = (now or ctx.clock()).astimezone(UTC) if ctx.kill_switch.is_armed(): await ctx.alert_manager.low( source="monitor_cycle", message="kill switch armed — skipping" ) return MonitorCycleResult(outcomes=[]) # Refresh spot+DVOL snapshot first; it goes into dvol_history so the # next cycle can compute return_4h. spot_t = asyncio.create_task(ctx.deribit.index_price_eth()) dvol_t = asyncio.create_task(ctx.deribit.latest_dvol(currency="ETH", now=when)) return_4h_t = asyncio.create_task(_fetch_return_4h(ctx, now=when)) await asyncio.gather(spot_t, dvol_t, return_4h_t) spot = spot_t.result() dvol = dvol_t.result() return_4h = return_4h_t.result() conn = connect_state(ctx.db_path) try: with transaction(conn): ctx.repository.record_dvol_snapshot( conn, DvolSnapshot(timestamp=when, asset="ETH", dvol=dvol, spot=spot), ) positions = ctx.repository.list_positions(conn, status="open") finally: conn.close() outcomes: list[PositionOutcome] = [] for record in positions: outcome = await _evaluate_position( ctx, record=record, spot=spot, dvol=dvol, return_4h=return_4h, now=when, ) outcomes.append(outcome) return MonitorCycleResult(outcomes=outcomes) async def _evaluate_position( ctx: RuntimeContext, *, record: PositionRecord, spot: Decimal, dvol: Decimal, return_4h: Decimal, now: datetime, ) -> PositionOutcome: snapshot = await _build_position_snapshot( ctx, record=record, spot=spot, dvol=dvol, return_4h=return_4h, eth_price_usd_now=spot, now=now, ) if snapshot is None: await ctx.alert_manager.high( source="monitor_cycle", message=( f"missing tickers for legs of position {record.proposal_id} — " "broker may have de-listed an instrument" ), ) return PositionOutcome( proposal_id=str(record.proposal_id), action="HOLD", closed=False, reason="missing_ticker", ) decision = evaluate(snapshot, ctx.cfg) conn = connect_state(ctx.db_path) try: with transaction(conn): ctx.repository.record_decision( conn, DecisionRecord( decision_type="exit_check", timestamp=now, inputs_json=json.dumps( { "spot": str(spot), "dvol": str(dvol), "mark_combo_now_eth": str(snapshot.mark_combo_now_eth), "delta_short_now": str(snapshot.delta_short_now), "return_4h_now": str(return_4h), }, default=str, sort_keys=True, ), outputs_json=json.dumps( { "action": decision.action, "reason": decision.reason, "pnl_eth": str(decision.pnl_estimate_eth), "pnl_usd": str(decision.pnl_estimate_usd), }, default=str, sort_keys=True, ), action_taken=decision.action, proposal_id=record.proposal_id, ), ) finally: conn.close() if decision.action == "HOLD": ctx.audit_log.append( event="HOLD", payload={ "proposal_id": str(record.proposal_id), "reason": decision.reason, }, now=now, ) return PositionOutcome( proposal_id=str(record.proposal_id), action="HOLD", closed=False, reason=decision.reason, ) # Place inverse combo to close. short_leg = ComboLegOrder( instrument_name=record.short_instrument, direction="buy" ) long_leg = ComboLegOrder( instrument_name=record.long_instrument, direction="sell" ) conn = connect_state(ctx.db_path) try: with transaction(conn): ctx.repository.update_position_status( conn, record.proposal_id, status="closing", now=now, ) finally: conn.close() try: order = await ctx.deribit.place_combo_order( legs=[short_leg, long_leg], side="buy", n_contracts=record.n_contracts, limit_price_eth=snapshot.mark_combo_now_eth / Decimal(record.n_contracts), label=f"bite-close-{record.proposal_id}", ) except Exception as exc: conn = connect_state(ctx.db_path) try: with transaction(conn): ctx.repository.update_position_status( conn, record.proposal_id, status="open", now=now ) finally: conn.close() await ctx.alert_manager.critical( source="monitor_cycle", message=( f"close failed for {record.proposal_id}: " f"{type(exc).__name__}: {exc}" ), component="runtime.monitor_cycle", ) return PositionOutcome( proposal_id=str(record.proposal_id), action=decision.action, closed=False, reason=f"close_error:{type(exc).__name__}", ) instruction_id = uuid4() is_filled = order.state in {"filled", "open"} next_status = "closed" if is_filled else "open" conn = connect_state(ctx.db_path) try: with transaction(conn): ctx.repository.create_instruction( conn, InstructionRecord( instruction_id=instruction_id, proposal_id=record.proposal_id, kind="close_combo", payload_json=json.dumps(order.raw, default=str, sort_keys=True), sent_at=now, actual_fill_eth=order.average_price_eth, ), ) ctx.repository.update_position_status( conn, record.proposal_id, status=next_status, # type: ignore[arg-type] closed_at=now if is_filled else None, close_reason=decision.action if is_filled else None, debit_paid_eth=order.average_price_eth if is_filled else None, pnl_eth=decision.pnl_estimate_eth if is_filled else None, pnl_usd=decision.pnl_estimate_usd if is_filled else None, now=now, ) finally: conn.close() if not is_filled: await ctx.alert_manager.critical( source="monitor_cycle", message=( f"close rejected by broker for {record.proposal_id} " f"(state={order.state})" ), component="runtime.monitor_cycle", ) return PositionOutcome( proposal_id=str(record.proposal_id), action=decision.action, closed=False, reason=f"broker_reject:{order.state}", ) ctx.audit_log.append( event="EXIT_FILLED", payload={ "proposal_id": str(record.proposal_id), "action": decision.action, "pnl_usd": str(decision.pnl_estimate_usd), "combo_instrument": order.combo_instrument, }, now=now, ) await ctx.telegram.notify_position_closed( instrument=order.combo_instrument, realized_pnl_usd=decision.pnl_estimate_usd, reason=decision.action, ) _log.info( "exit filled: proposal=%s action=%s pnl_usd=%s", record.proposal_id, decision.action, decision.pnl_estimate_usd, ) return PositionOutcome( proposal_id=str(record.proposal_id), action=decision.action, closed=True, reason=decision.reason, )