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..() got an unexpected keyword argument 'account_nick' rtrader/cli/trader_liquidation.py:2445: TypeError