| monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x1201a4800>
|
|
|
| def test_liquidation_orchestrator_creates_review_plan(monkeypatch):
|
| fake_broker = FakeBroker()
|
| created_plans: List[Dict[str, Any]] = []
|
|
|
| monkeypatch.setattr(trader_liquidation.PlanBrokerRegistry, "get", lambda mode: fake_broker)
|
| monkeypatch.setattr(trader_liquidation, "QuoteThrottleService", FakeQuoteService)
|
| monkeypatch.setattr(trader_liquidation, "_existing_liquidation_plan", lambda symbol, trade_day: False)
|
| monkeypatch.setattr(
|
| trader_liquidation.scheduler,
|
| "trading_day",
|
| lambda _mode=None: date(2025, 1, 2),
|
| )
|
| monkeypatch.setattr(
|
| trader_liquidation.scheduler,
|
| "is_trading_day",
|
| lambda trade_day: True,
|
| )
|
| monkeypatch.setattr(
|
| trader_liquidation,
|
| "add_trading_plan",
|
| lambda **kwargs: created_plans.append(kwargs),
|
| )
|
| monkeypatch.setattr(trader_liquidation, "TraderLiquidationAuditService", lambda: DummyAuditService())
|
| monkeypatch.setattr(trader_liquidation, "TraderEODRepository", lambda *args, **kwargs: DummyAuditRepository())
|
| monkeypatch.setattr(trader_liquidation.scheduler, "is_trading_hour", lambda: True)
|
| monkeypatch.setattr(trader_liquidation.scheduler, "in_premarket_window", lambda: False)
|
| monkeypatch.setattr(trader_liquidation.scheduler, "market_opened", lambda: 30)
|
| monkeypatch.setattr(trader_liquidation.scheduler, "market_to_open", lambda: 0)
|
| monkeypatch.setattr(trader_liquidation.scheduler, "market_remain", lambda: 240)
|
|
|
| > trader_liquidation.run_liquidation_orchestrator(
|
| alpaca_mode="paper",
|
| poll_interval=1,
|
| max_cycles=1,
|
| dry_run=False,
|
| )
|
|
|
| tests/test_trader_liquidation_orchestrator.py:106:
|
| _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
|
|
|
| def run_liquidation_orchestrator(
|
| *,
|
| broker_backend: str = "alpaca",
|
| account_nick: Optional[str] = None,
|
| alpaca_mode: Optional[str] = None,
|
| poll_interval: int = 60,
|
| max_cycles: Optional[int] = None,
|
| default_stop_loss_pct: Optional[float] = None,
|
| dry_run: bool = False,
|
| skip_signal_handlers: bool = False,
|
| ) -> None:
|
| """
|
| Continuously monitor live positions and stage review sell plans when liquidation
|
| heuristics trigger.
|
|
|
| Designed for long-running PM2 jobs; exits after ``max_cycles`` when provided.
|
| """
|
| fallback_job_name: Optional[str] = None
|
| broker_backend_norm = (broker_backend or "").strip().lower()
|
| if broker_backend_norm == "alpaca" and account_nick:
|
| fallback_job_name = f"trader-liquidation-alpaca-{account_nick}"
|
| elif broker_backend_norm:
|
| # NOTE: Alpaca PM2 entries are account-specific; if account_nick is missing, there is no
|
| # corresponding PM2 name to synthesize. In that case, rely on PM2_PROCESS_NAME/PM2_NAME
|
| # when available, and only use this fallback when running outside PM2.
|
| fallback_job_name = (
|
| "trader-liquidation-alpaca"
|
| if broker_backend_norm == "alpaca"
|
| else f"trader-liquidation-orchestrator-{broker_backend_norm}"
|
| )
|
|
|
| # Send startup notification (skip for account-monitor threads).
|
| if not skip_signal_handlers:
|
| # Prefer PM2's real process name when running under PM2; fallback is for direct invocation.
|
| notify_pm2_job_startup(fallback=fallback_job_name)
|
|
|
| # Track if we should exit
|
| _shutdown_requested = {"stop": False}
|
|
|
| def _coerce_float(value: Any, default: float) -> float:
|
| try:
|
| return float(value)
|
| except (TypeError, ValueError):
|
| return default
|
|
|
| def _coerce_int(value: Any, default: int) -> int:
|
| try:
|
| return int(value)
|
| except (TypeError, ValueError):
|
| return default
|
|
|
| def _coerce_bool(value: Any, default: bool) -> bool:
|
| if value is None:
|
| return default
|
| if isinstance(value, bool):
|
| return value
|
| if isinstance(value, str):
|
| normalized = value.strip().lower()
|
| if normalized in {"1", "true", "yes", "on"}:
|
| return True
|
| if normalized in {"0", "false", "no", "off"}:
|
| return False
|
| return bool(value)
|
|
|
| def _flow_signal_from_result(result: FlowEvaluationResult) -> FlowSignal:
|
| metadata: Dict[str, Any] = {}
|
| matched = result.matched_window
|
| if matched is not None:
|
| metadata["max_delta_pct"] = round(float(matched.max_delta or 0.0) * 100.0, 4)
|
| metadata["window_span_indexes"] = [matched.start_index, matched.end_index]
|
| if result.last_60_delta_ratio is not None:
|
| metadata["last_60_delta_pct"] = round(float(result.last_60_delta_ratio) * 100.0, 4)
|
| if result.opening_delta_ratio is not None:
|
| metadata["opening_delta_pct"] = round(float(result.opening_delta_ratio) * 100.0, 4)
|
| if result.cmf_value is not None:
|
| metadata["cmf_value"] = round(float(result.cmf_value), 4)
|
| return FlowSignal(
|
| bias=result.bias,
|
| bias_met=result.bias_met,
|
| delta_ratio=result.delta_ratio,
|
| buyer_volume=result.buyer_volume,
|
| seller_volume=result.seller_volume,
|
| window_minutes=result.window_minutes,
|
| window_start=result.window_start,
|
| window_end=result.window_end,
|
| metadata=metadata,
|
| )
|
|
|
| broker_backend_normalized = str(broker_backend or "alpaca").strip().lower()
|
| if broker_backend_normalized not in {"alpaca", "robinhood", "public"}:
|
| raise ValueError(f"Unsupported broker backend '{broker_backend}'")
|
|
|
| if broker_backend_normalized == "alpaca":
|
| normalized_mode = normalize_alpaca_mode(alpaca_mode or settings.get("ALPACA_MODE", "live"))
|
| plan_mode = normalized_mode
|
| elif broker_backend_normalized == "robinhood":
|
| normalized_mode = "robinhood"
|
| plan_mode = None
|
| else:
|
| normalized_mode = "public"
|
| plan_mode = None
|
|
|
| if broker_backend_normalized == "alpaca" and account_nick:
|
| try:
|
| from rtrader.brokers.alpaca import Alpaca # noqa: PLC0415 - runtime import to avoid heavy dependency
|
|
|
| account_cfg = Alpaca._resolve_account_config(account_nick) # type: ignore[attr-defined]
|
| except Exception:
|
| account_cfg = None
|
| if not account_cfg:
|
| raise ValueError(f"Alpaca account nickname '{account_nick}' not found in settings.")
|
|
|
|
|
| target_gain_pct = _coerce_float(settings.get("LIQUIDATION_TARGET_GAIN_PCT", 1.0), 1.0)
|
| trailing_stop_pct = _coerce_float(settings.get("LIQUIDATION_TRAILING_STOP_GIVEBACK_PCT", 0.5), 0.5)
|
| default_flow_delta_ratio = _coerce_float(settings.get("BUYER_SELLER_FLOW_MIN_DELTA_PCT", 0.005), 0.005)
|
| configured_flow_delta_pct = settings.get("LIQUIDATION_FLOW_DELTA_PCT", None)
|
| if configured_flow_delta_pct is None:
|
| flow_delta_pct = default_flow_delta_ratio * 100.0
|
| else:
|
| flow_delta_pct = _coerce_float(configured_flow_delta_pct, default_flow_delta_ratio * 100.0)
|
| flow_window_minutes = _coerce_int(
|
| settings.get(
|
| "LIQUIDATION_FLOW_WINDOW_MINUTES",
|
| settings.get("BUYER_SELLER_FLOW_WINDOW_MINUTES", 60),
|
| ),
|
| 60,
|
| )
|
| flow_bias = str(settings.get("LIQUIDATION_FLOW_BIAS", "seller") or "seller").lower()
|
| require_flow_confirmation = _coerce_bool(
|
| settings.get("LIQUIDATION_REQUIRE_FLOW_CONFIRMATION", True),
|
| True,
|
| )
|
| # Auto-sell bias threshold: seller must exceed buyer by this % to trigger auto-sell
|
| # Default 3.0% is less sensitive than the old hardcoded 0.5%
|
| auto_sell_bias_threshold_pct = _coerce_float(
|
| settings.get("AUTO_SELL_BIAS_THRESHOLD_PCT", 3.0), 3.0
|
| )
|
| cmf_divergence_min_gain_pct = _coerce_float(settings.get("LIQUIDATION_CMF_DIVERGENCE_MIN_GAIN_PCT", 0.5), 0.5)
|
| cmf_divergence_min_drawdown_pct = _coerce_float(settings.get("LIQUIDATION_CMF_DIVERGENCE_MIN_DRAWDOWN_PCT", 0.5), 0.5)
|
| cmf_divergence_min_slope = _coerce_float(settings.get("LIQUIDATION_CMF_DIVERGENCE_MIN_SLOPE", 0.02), 0.02)
|
| cmf_divergence_relative_min_slope = _coerce_float(settings.get("LIQUIDATION_CMF_DIVERGENCE_RELATIVE_MIN_SLOPE", 0.02), 0.02)
|
| cmf_divergence_relative_gap = _coerce_float(settings.get("LIQUIDATION_CMF_DIVERGENCE_RELATIVE_GAP", 0.05), 0.05)
|
| cmf_divergence_value_floor = _coerce_float(settings.get("LIQUIDATION_CMF_DIVERGENCE_VALUE_FLOOR", 0.05), 0.05)
|
| cmf_divergence_hold_minutes = _coerce_int(settings.get("LIQUIDATION_CMF_DIVERGENCE_HOLD_MINUTES", 45), 45)
|
| opening_hold_window = settings.get(
|
| "LIQUIDATION_OPENING_HOLD_WINDOW_MINUTES",
|
| 15.0,
|
| )
|
| closing_block_minutes = _coerce_int(settings.get("LIQUIDATION_CLOSING_BLOCK_MINUTES", 15), 15)
|
|
|
| # Velocity divergence settings (multi-day price-CMF divergence detection)
|
| velocity_divergence_enabled = _coerce_bool(
|
| settings.get("LIQUIDATION_VELOCITY_DIVERGENCE_ENABLED", False), False
|
| )
|
| velocity_divergence_lookback_days = _coerce_int(
|
| settings.get("LIQUIDATION_VELOCITY_DIVERGENCE_LOOKBACK_DAYS", 3), 3
|
| )
|
| velocity_divergence_ratio_threshold = _coerce_float(
|
| settings.get("LIQUIDATION_VELOCITY_DIVERGENCE_RATIO_THRESHOLD", 2.0), 2.0
|
| )
|
| velocity_divergence_cumulative_threshold = _coerce_float(
|
| settings.get("LIQUIDATION_VELOCITY_DIVERGENCE_CUMULATIVE_THRESHOLD", 3.0), 3.0
|
| )
|
| velocity_divergence_min_price_gain_pct = _coerce_float(
|
| settings.get("LIQUIDATION_VELOCITY_DIVERGENCE_MIN_PRICE_GAIN_PCT", 2.0), 2.0
|
| )
|
| velocity_divergence_min_confidence = _coerce_float(
|
| settings.get("LIQUIDATION_VELOCITY_DIVERGENCE_MIN_CONFIDENCE", 0.6), 0.6
|
| )
|
|
|
| # Sector ETF divergence settings
|
| sector_divergence_enabled = _coerce_bool(
|
| settings.get("LIQUIDATION_SECTOR_DIVERGENCE_ENABLED", True), True
|
| )
|
| sector_divergence_threshold = _coerce_float(
|
| settings.get("LIQUIDATION_SECTOR_DIVERGENCE_THRESHOLD", 0.02), 0.02
|
| )
|
| sector_divergence_sell_threshold = _coerce_float(
|
| settings.get("LIQUIDATION_SECTOR_DIVERGENCE_SELL_THRESHOLD", 0.03), 0.03
|
| )
|
| sector_divergence_min_gain_pct = _coerce_float(
|
| settings.get("LIQUIDATION_SECTOR_DIVERGENCE_MIN_GAIN_PCT", 0.5), 0.5
|
| )
|
| sector_divergence_buy_enabled = _coerce_bool(
|
| settings.get("LIQUIDATION_SECTOR_DIVERGENCE_BUY_ENABLED", False), False
|
| )
|
|
|
| liquidation_engine = LiquidationDecisionEngine(
|
| target_gain_pct=target_gain_pct,
|
| trailing_stop_pct=trailing_stop_pct,
|
| trailing_stop_enabled=bool(settings.get("TRAILING_STOP_ENABLED", True)),
|
| min_flow_delta_pct=flow_delta_pct,
|
| flow_bias=flow_bias,
|
| require_flow_confirmation=require_flow_confirmation,
|
| cmf_divergence_min_gain_pct=cmf_divergence_min_gain_pct,
|
| cmf_divergence_min_drawdown_pct=cmf_divergence_min_drawdown_pct,
|
| cmf_divergence_min_slope=cmf_divergence_min_slope,
|
| cmf_divergence_relative_min_slope=cmf_divergence_relative_min_slope,
|
| cmf_divergence_relative_gap=cmf_divergence_relative_gap,
|
| cmf_divergence_value_floor=cmf_divergence_value_floor,
|
| cmf_divergence_hold_minutes=cmf_divergence_hold_minutes,
|
| prevent_liquidation_window=opening_hold_window,
|
| )
|
| flow_service = CMFService()
|
|
|
| account_label = f" | account={account_nick}" if (broker_backend_normalized == "alpaca" and account_nick) else ""
|
| logger.info(
|
| "[TraderLiquidation] Initialising orchestrator | backend={} | mode={} | poll_interval={}s | max_cycles={} | dry_run={}{}",
|
| broker_backend_normalized,
|
| normalized_mode,
|
| poll_interval,
|
| max_cycles,
|
| dry_run,
|
| account_label,
|
| )
|
| audit_service = TraderLiquidationAuditService()
|
| audit_repository = TraderEODRepository(workflow=TRADER_LIQUIDATION_AUDIT_WORKFLOW)
|
| audit_flags = {
|
| "broker_backend": broker_backend_normalized,
|
| "alpaca_mode": plan_mode,
|
| "account_nick": account_nick,
|
| "poll_interval": poll_interval,
|
| "max_cycles": max_cycles,
|
| "default_stop_loss_pct": default_stop_loss_pct,
|
| "dry_run": dry_run,
|
| "target_gain_pct": target_gain_pct,
|
| "trailing_stop_pct": trailing_stop_pct,
|
| "flow_delta_pct": flow_delta_pct,
|
| "flow_window_minutes": flow_window_minutes,
|
| "flow_bias": flow_bias,
|
| "require_flow_confirmation": require_flow_confirmation,
|
| "auto_sell_bias_threshold_pct": auto_sell_bias_threshold_pct,
|
| "cmf_divergence_min_gain_pct": cmf_divergence_min_gain_pct,
|
| "cmf_divergence_min_drawdown_pct": cmf_divergence_min_drawdown_pct,
|
| "cmf_divergence_min_slope": cmf_divergence_min_slope,
|
| "cmf_divergence_relative_min_slope": cmf_divergence_relative_min_slope,
|
| "cmf_divergence_relative_gap": cmf_divergence_relative_gap,
|
| "cmf_divergence_value_floor": cmf_divergence_value_floor,
|
| "cmf_divergence_hold_minutes": cmf_divergence_hold_minutes,
|
| "opening_hold_window_minutes": opening_hold_window,
|
| }
|
| current_run: Optional[Dict[str, Any]] = None
|
|
|
| def _update_heartbeat(status: str, message: str) -> None:
|
| try:
|
| update_job_heartbeat(LIQUIDATION_HEARTBEAT_JOB, status=status, message=message)
|
| except Exception as exc: # pragma: no cover - heartbeat best-effort
|
| logger.debug("Trader liquidation heartbeat update failed: {}", exc)
|
|
|
| _update_heartbeat(
|
| "running",
|
| f"Initializing orchestrator | backend={broker_backend_normalized} mode={normalized_mode}"
|
| f"{f' account={account_nick}' if (broker_backend_normalized == 'alpaca' and account_nick) else ''} "
|
| f"dry_run={dry_run} poll={poll_interval}s",
|
| )
|
|
|
| def ensure_run(trading_day: date) -> None:
|
| nonlocal current_run
|
| if current_run and current_run["trading_day"] == trading_day:
|
| return
|
| if current_run:
|
| finalize_current_run("success", skip_reason="trading_day_rollover")
|
|
|
| run_id = uuid.uuid4().hex[:12]
|
| attempt = 1
|
| started_at = datetime.now(timezone.utc)
|
| try:
|
| audit_run = audit_service.begin_run(trading_day=trading_day, run_id=run_id, flags=audit_flags)
|
| attempt = getattr(audit_run, "attempt", attempt)
|
| started_at = getattr(audit_run, "started_at", started_at)
|
| except Exception as exc: # pragma: no cover - defensive
|
| logger.warning(
|
| "[TraderLiquidation] Audit run initialisation failed | day={} | run_id={} | error={}",
|
| trading_day,
|
| run_id,
|
| exc,
|
| )
|
|
|
| audit_meta = {
|
| "run_id": run_id,
|
| "attempt": attempt,
|
| "cli_flags": dict(audit_flags),
|
| "started_at": started_at,
|
| }
|
| summary: Dict[str, Any] = {
|
| "total_cycles": 0,
|
| "evaluated_positions": 0,
|
| "plans_created": 0,
|
| "skipped_existing": 0,
|
| "locked_positions": 0,
|
| "market_closed_cycles": 0,
|
| "dry_run": dry_run,
|
| "broker_backend": broker_backend_normalized,
|
| "alpaca_mode": plan_mode,
|
| "account_nick": account_nick,
|
| "max_cycles": max_cycles,
|
| "poll_interval": poll_interval,
|
| "default_stop_loss_pct": default_stop_loss_pct,
|
| "sell_events": [],
|
| "sell_events_count": 0,
|
| "flow_confirmed_count": 0,
|
| }
|
| audit_meta["liquidation_summary"] = dict(summary)
|
| current_run = {
|
| "trading_day": trading_day,
|
| "run_id": run_id,
|
| "attempt": attempt,
|
| "audit_meta": audit_meta,
|
| "summary": summary,
|
| }
|
| logger.info(
|
| "[TraderLiquidation] Audit run started | trading_day={} | run_id={} | attempt={}",
|
| trading_day,
|
| run_id,
|
| attempt,
|
| )
|
| try:
|
| audit_service.update_run_fields(
|
| run_id,
|
| {
|
| "status": "running",
|
| "metadata": audit_meta,
|
| "trading_day": trading_day.isoformat(),
|
| },
|
| )
|
| except Exception as exc: # pragma: no cover - best effort
|
| logger.debug(
|
| "[TraderLiquidation] Unable to persist initial run metadata | run_id={} | error={}",
|
| run_id,
|
| exc,
|
| )
|
| _update_heartbeat(
|
| "running",
|
| f"Run {run_id} started | day={trading_day.isoformat()} | attempt={attempt}"
|
| f"{account_label}",
|
| )
|
|
|
| def finalize_current_run(
|
| status: str,
|
| *,
|
| error_message: Optional[str] = None,
|
| skip_reason: Optional[str] = None,
|
| ) -> None:
|
| nonlocal current_run
|
| if not current_run:
|
| return
|
|
|
| summary = dict(current_run["summary"])
|
| summary["status"] = status
|
| summary["completed_cycles"] = summary.get("total_cycles", 0)
|
| total_cycles = summary.get("total_cycles") or 0
|
| positions_total = summary.get("positions_observed_total") or 0
|
| if total_cycles > 0 and positions_total:
|
| summary["positions_observed_avg"] = round(positions_total / total_cycles, 2)
|
| if summary.get("warnings"):
|
| summary["warnings"] = list(dict.fromkeys(str(item) for item in summary["warnings"]))
|
| if summary.get("errors"):
|
| summary["errors"] = list(dict.fromkeys(str(item) for item in summary["errors"]))
|
| audit_meta = dict(current_run["audit_meta"])
|
| audit_meta["liquidation_summary"] = summary
|
| audit_meta["liquidation_flags"] = dict(audit_flags)
|
| current_run["audit_meta"] = audit_meta
|
| context_config = {
|
| "audit": audit_meta,
|
| "liquidation_summary": summary,
|
| "liquidation_flags": dict(audit_flags),
|
| }
|
| try:
|
| audit_service.finalize_run(
|
| trading_day=current_run["trading_day"],
|
| run_id=current_run["run_id"],
|
| status=status,
|
| context_config=context_config,
|
| error_message=error_message,
|
| skip_reason=skip_reason,
|
| )
|
| except Exception as exc: # pragma: no cover - defensive
|
| logger.warning(
|
| "[TraderLiquidation] Audit finalisation failed | day={} | run_id={} | error={}",
|
| current_run["trading_day"],
|
| current_run["run_id"],
|
| exc,
|
| )
|
| status_lower = (status or "").lower()
|
| heartbeat_status = {
|
| "success": "completed",
|
| "completed": "completed",
|
| "skipped": "idle",
|
| "failed": "error",
|
| "error": "error",
|
| }.get(status_lower, status_lower or "completed")
|
| heartbeat_message = (
|
| f"{status_lower or 'completed'} run {current_run['run_id']} | day={current_run['trading_day'].isoformat()} "
|
| f"cycles={summary.get('total_cycles', 0)} plans={summary.get('plans_created', 0)}"
|
| )
|
| if skip_reason:
|
| heartbeat_message += f" | skip={skip_reason}"
|
| if error_message:
|
| heartbeat_message += f" | error={error_message}"
|
| _update_heartbeat(heartbeat_status, heartbeat_message)
|
| current_run = None
|
|
|
| def record_cycle(
|
| trading_day: date,
|
| cycle_index: int,
|
| status: str,
|
| message: str,
|
| payload: Dict[str, Any],
|
| *,
|
| error: Optional[str] = None,
|
| summary_updates: Optional[Dict[str, int]] = None,
|
| ) -> None:
|
| nonlocal current_run
|
| if not current_run:
|
| return
|
|
|
| now = datetime.now(timezone.utc)
|
| task_doc = {
|
| "task_id": uuid.uuid4().hex,
|
| "step": f"cycle/{cycle_index:05d}",
|
| "status": status,
|
| "payload_out": {"message": message, **payload},
|
| "error": error,
|
| "started_at": now,
|
| "completed_at": now,
|
| "created_at": now,
|
| "trading_day": trading_day,
|
| "run_id": current_run["run_id"],
|
| "run_attempt": current_run["attempt"],
|
| }
|
| try:
|
| audit_repository.save_task(task_doc)
|
| except Exception as exc: # pragma: no cover - best effort
|
| logger.warning(
|
| "[TraderLiquidation] Audit task persistence failed | run_id={} | cycle={} | error={}",
|
| current_run["run_id"],
|
| cycle_index,
|
| exc,
|
| )
|
|
|
| summary = current_run["summary"]
|
| summary["total_cycles"] = summary.get("total_cycles", 0) + 1
|
| status_counts = summary.setdefault("status_counts", {})
|
| status_counts[status] = status_counts.get(status, 0) + 1
|
|
|
| if summary_updates:
|
| for key, value in summary_updates.items():
|
| if value is None:
|
| continue
|
| summary[key] = summary.get(key, 0) + int(value)
|
|
|
| warnings = payload.get("warnings") or []
|
| if warnings:
|
| warning_list = summary.setdefault("warnings", [])
|
| warning_list.extend(str(item) for item in warnings)
|
|
|
| sell_events = payload.get("sell_events") or []
|
| if sell_events:
|
| sell_store = summary.setdefault("sell_events", [])
|
| sell_store.extend(sell_events)
|
| if len(sell_store) > 200:
|
| summary["sell_events"] = sell_store[-200:]
|
|
|
| if error:
|
| error_list = summary.setdefault("errors", [])
|
| error_list.append(str(error))
|
|
|
| summary["last_cycle"] = {
|
| "cycle": cycle_index,
|
| "status": status,
|
| "timestamp": now.isoformat(),
|
| **{key: value for key, value in payload.items() if key != "warnings"},
|
| }
|
|
|
| current_run["audit_meta"]["liquidation_summary"] = dict(summary)
|
| try:
|
| audit_service.update_run_fields(
|
| current_run["run_id"],
|
| {
|
| "status": "running",
|
| "metadata": current_run["audit_meta"],
|
| },
|
| )
|
| except Exception as exc: # pragma: no cover - best effort
|
| logger.debug(
|
| "[TraderLiquidation] Unable to update run metadata | run_id={} | cycle={} | error={}",
|
| current_run["run_id"],
|
| cycle_index,
|
| exc,
|
| )
|
|
|
| activity_entry = {
|
| "step": task_doc["step"],
|
| "status": status,
|
| "started_at": task_doc["started_at"].isoformat(),
|
| "completed_at": task_doc["completed_at"].isoformat(),
|
| "message": payload.get("message") or "",
|
| "reason": payload.get("reason"),
|
| "cycle": cycle_index,
|
| "evaluated_positions": payload.get("evaluated_positions"),
|
| "plans_created": payload.get("plans_created"),
|
| "skipped_existing": payload.get("skipped_existing"),
|
| "positions_observed": payload.get("positions_observed"),
|
| "warnings": payload.get("warnings") or [],
|
| "sell_events": payload.get("sell_events") or [],
|
| "position_decisions": payload.get("position_decisions") or [],
|
| "minutes_since_open": payload.get("minutes_since_open"),
|
| "minutes_to_open": payload.get("minutes_to_open"),
|
| "market_status": payload.get("market_status"),
|
| "in_regular_hours": payload.get("in_regular_hours"),
|
| "in_premarket": payload.get("in_premarket"),
|
| "dry_run": payload.get("dry_run"),
|
| "outside_market_hours": bool(
|
| payload.get("outside_market_hours")
|
| or payload.get("reason") == "outside_market_hours"
|
| ),
|
| "outside_trading_day": bool(
|
| payload.get("outside_trading_day")
|
| or payload.get("reason") == "outside_trading_day"
|
| ),
|
| }
|
|
|
| summary_payload = {
|
| "total_cycles": summary.get("total_cycles"),
|
| "market_closed_cycles": summary.get("market_closed_cycles"),
|
| "plans_created": summary.get("plans_created"),
|
| "skipped_existing": summary.get("skipped_existing"),
|
| "evaluated_positions": summary.get("evaluated_positions"),
|
| "positions_observed_total": summary.get("positions_observed_total"),
|
| "positions_observed_avg": summary.get("positions_observed_avg"),
|
| "sell_events_count": summary.get("sell_events_count"),
|
| "flow_confirmed_count": summary.get("flow_confirmed_count"),
|
| "warnings": summary.get("warnings"),
|
| "errors": summary.get("errors"),
|
| }
|
|
|
| try:
|
| LiquidationEventPublisher.publish_cycle(
|
| trading_day=trading_day,
|
| run_id=current_run["run_id"],
|
| cycle=cycle_index,
|
| summary=summary_payload,
|
| activity=activity_entry,
|
| )
|
| except Exception as exc: # pragma: no cover - defensive logging
|
| logger.debug(
|
| "[TraderLiquidation] Unable to publish SSE event | run_id={} | cycle={} | error={}",
|
| current_run["run_id"],
|
| cycle_index,
|
| exc,
|
| )
|
|
|
| heartbeat_message = (
|
| f"{status.upper()} cycle {cycle_index} | day={trading_day.isoformat()} "
|
| f"cycles={summary.get('total_cycles', 0)} plans={summary.get('plans_created', 0)} "
|
| f"evaluated={summary.get('evaluated_positions', 0)} skipped_existing={summary.get('skipped_existing', 0)}"
|
| )
|
| heartbeat_message = (
|
| f"{heartbeat_message} | backend={broker_backend_normalized} mode={normalized_mode}"
|
| )
|
| if broker_backend_normalized == "alpaca" and account_nick:
|
| heartbeat_message = f"{heartbeat_message} account={account_nick}"
|
| reason_token = payload.get("reason") or payload.get("message")
|
| if reason_token:
|
| heartbeat_message += f" | reason={reason_token}"
|
| if error:
|
| heartbeat_message += f" | error={error}"
|
| _update_heartbeat("running", heartbeat_message)
|
|
|
| if broker_backend_normalized == "alpaca":
|
| try:
|
| > broker = PlanBrokerRegistry.get(normalized_mode, account_nick=account_nick)
|
| ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
|
| E TypeError: test_liquidation_orchestrator_creates_review_plan.<locals>.<lambda>() got an unexpected keyword argument 'account_nick'
|
|
|
| rtrader/cli/trader_liquidation.py:2445: TypeError
|