diff --git a/app/trading/practice_trader.py b/app/trading/practice_trader.py index 61d0e27..21900d1 100644 --- a/app/trading/practice_trader.py +++ b/app/trading/practice_trader.py @@ -911,6 +911,24 @@ def _compact_account_state_json(state: Mapping[str, Any]) -> dict[str, Any]: return compacted +def _write_state_file_atomically(payload: Mapping[str, Any]) -> None: + """Replace the canonical account JSON without exposing a partial file.""" + tmp = STATE_FILE.with_name( + f"{STATE_FILE.name}.{os.getpid()}.{threading.get_ident()}.tmp" + ) + try: + tmp.write_text( + json.dumps(payload, ensure_ascii=False, indent=2), + encoding="utf-8", + ) + tmp.replace(STATE_FILE) + finally: + try: + tmp.unlink(missing_ok=True) + except OSError: + pass + + _STATE_FILE_THREAD_LOCK = threading.RLock() _STATE_FILE_LOCK_DEPTH = threading.local() @@ -1389,12 +1407,13 @@ def trade_id(item: dict[str, Any]) -> tuple[str, ...]: if current_market_time > state_market_time: state["market_decision_context"] = current_market_ctx - unarchived_history = { - kind: list(state.get(kind) or []) - for kind in JSON_RECENT_HISTORY_LIMITS - if isinstance(state.get(kind), list) - } - history_archived = _archive_account_history_before_compaction(state) + # Commit the complete account state before writing any SQLite projection. + # If the replace fails (for example because a bind-mounted state file is + # owned by root), no trade, decision, or history row may become visible. + full_state = dict(state) + _write_state_file_atomically(full_state) + + history_archived = _archive_account_history_before_compaction(full_state) prune_non_trading_day_equity_points(state) prune_future_intraday_equity_points(state) @@ -1403,11 +1422,16 @@ def trade_id(item: dict[str, Any]) -> tuple[str, ...]: if history_archived: persisted_state = _compact_account_state_json(state) - else: - # Existing JSON remains the recovery source until a complete archive - # transaction succeeds, including rows rejected by active-view cleanup. - persisted_state = dict(state) - persisted_state.update(unarchived_history) + try: + _write_state_file_atomically(persisted_state) + except OSError as exc: + # The full JSON above is already the committed recovery source. + # Compaction is optional and can be retried by a later save. + print( + "[WARN] 账户状态已保存，但 JSON 压缩失败，保留完整历史: " + f"{type(exc).__name__}", + flush=True, + ) equity_point_to_sync = next( ( @@ -1418,13 +1442,6 @@ def trade_id(item: dict[str, Any]) -> tuple[str, ...]: ), None, ) - tmp = STATE_FILE.with_name(f"{STATE_FILE.name}.{os.getpid()}.{threading.get_ident()}.tmp") - tmp.write_text( - json.dumps(persisted_state, ensure_ascii=False, indent=2), - encoding="utf-8", - ) - tmp.replace(STATE_FILE) - if rejected_trade_count: # A stale writer may already have replaced today's mutable SQLite # position snapshot before its oversell was rejected during merge. @@ -7207,13 +7224,24 @@ def _commit_refreshed_auto_exits( ) canonical_state.pop(AUTO_EXIT_PERSISTENCE_STATUS_KEY, None) canonical_state[AUTO_EXIT_ELIGIBLE_CODES_KEY] = sorted(eligible_codes) + decision_log_size = len(canonical_state.get("decision_log") or []) try: executed = check_auto_exits(canonical_state, dt) finally: canonical_state.pop(AUTO_EXIT_ELIGIBLE_CODES_KEY, None) - persistence_status = _pop_auto_exit_persistence_status(canonical_state) + new_decisions = [ + entry + for entry in (canonical_state.get("decision_log") or [])[decision_log_size:] + if isinstance(entry, dict) + ] + canonical_state.pop(AUTO_EXIT_PERSISTENCE_STATUS_KEY, None) record_equity(canonical_state) save_state(canonical_state) + persistence_status = _sync_committed_account_projections( + canonical_state, + trades=executed, + decisions=new_decisions, + ) return canonical_state, executed, persistence_status @@ -7481,9 +7509,6 @@ def check_auto_exits( state["cash"] = round(cash, 2) state.setdefault("trade_log", []).extend(executed) del state["trade_log"][:-TRADE_LOG_LIMIT] - trades_persisted = _sync_trades_to_db(executed) - if trades_persisted: - _sync_positions_to_db(state) # 记录系统自动退出决策 log_entry = { "time": now_ts(), @@ -7499,15 +7524,7 @@ def check_auto_exits( "executed": executed, } state.setdefault("decision_log", []).append(log_entry) - decision_persisted = _sync_decision_to_db(log_entry) - state[AUTO_EXIT_PERSISTENCE_STATUS_KEY] = { - "trades_persisted": trades_persisted, - "decision_persisted": decision_persisted, - "durable_evidence_persisted": ( - trades_persisted and decision_persisted - ), - } - + return executed @@ -11337,13 +11354,42 @@ def _decision_has_candidate_evidence(log_entry: Mapping[str, Any]) -> bool: def _sync_positions_to_db(state: dict[str, Any]): - """将当前持仓快照同步写入 SQLite。""" + """将最新规范持仓快照同步写入 SQLite。""" try: from niuniu_db import snapshot_positions as _sp - _sp(state.get("positions", {})) + + # A different writer may commit after this caller's save_state() and + # before its projection begins. Re-read while holding the same account + # lock so a delayed projection cannot replace SQLite with stale holdings. + with state_file_write_lock(): + canonical_state = load_state() if STATE_FILE.exists() else state + _sp(canonical_state.get("positions", {})) except Exception: pass +def _sync_committed_account_projections( + state: dict[str, Any], + *, + trades: list[dict[str, Any]] | None = None, + decisions: list[dict[str, Any]] | None = None, +) -> dict[str, bool]: + """Project only account events whose canonical state commit has succeeded.""" + trade_rows = [item for item in (trades or []) if isinstance(item, dict)] + decision_rows = [item for item in (decisions or []) if isinstance(item, dict)] + decision_results = [_sync_decision_to_db(item) for item in decision_rows] + decision_persisted = all(result is True for result in decision_results) + trades_persisted = _sync_trades_to_db(trade_rows) + if trade_rows and trades_persisted: + _sync_positions_to_db(state) + return { + "trades_persisted": trades_persisted, + "decision_persisted": decision_persisted, + "durable_evidence_persisted": ( + trades_persisted and decision_persisted + ), + } + + def record_decision_log_entry(log_entry: dict[str, Any], *, mark_b1_done: bool = False) -> None: """Append a visible practice decision/event log and sync it to SQLite.""" state = load_state() @@ -11355,8 +11401,8 @@ def record_decision_log_entry(log_entry: dict[str, Any], *, mark_b1_done: bool = state["last_error"] = log_entry["decision"]["error"] if mark_b1_done and generated_at: state["last_b1_generated_at"] = generated_at - _sync_decision_to_db(log_entry) save_state(state) + _sync_decision_to_db(log_entry) def _fallback_action_reason(action: dict[str, Any], candidate: dict[str, Any] | None, act: str, name: str) -> str: @@ -11645,6 +11691,7 @@ def execute_due_pending_decisions(now: datetime | None = None) -> dict[str, Any] return {"executed": [], "attempted": 0} all_executed: list[dict[str, Any]] = [] + committed_decisions: list[dict[str, Any]] = [] attempted = 0 changed = False for entry in pending: @@ -11738,18 +11785,26 @@ def execute_due_pending_decisions(now: datetime | None = None) -> dict[str, Any] state.setdefault("decision_log", []).append(log_entry) del state["decision_log"][:-50] state["last_decision_at"] = log_entry["time"] - _sync_decision_to_db(log_entry) + committed_decisions.append(log_entry) if changed: - if all_executed: - _sync_trades_to_db(all_executed) - _sync_positions_to_db(state) record_equity(state) save_state(state) + persistence_status = _sync_committed_account_projections( + state, + trades=all_executed, + decisions=committed_decisions, + ) all_executed = _accounted_trade_executions(all_executed) if all_executed: _notify_trade_executions_safely(all_executed) - return {"executed": all_executed, "attempted": attempted} + else: + persistence_status = _default_persistence_status() + return { + "executed": all_executed, + "attempted": attempted, + **persistence_status, + } def run_decision_after_b1(b1_payload: dict[str, Any], force: bool = False) -> dict[str, Any]: @@ -11789,13 +11844,13 @@ def run_decision_after_b1(b1_payload: dict[str, Any], force: bool = False) -> di str(generated_at), str(schedule_slot), ) + record_equity(state) + save_state(state) decision_persisted = bool( prior_decision and _decision_has_candidate_evidence(prior_decision) and _sync_decision_to_db(prior_decision) ) - record_equity(state) - save_state(state) if position_exit_executed: _notify_trade_executions_safely(position_exit_executed) return { @@ -12090,13 +12145,16 @@ def make_decision(reason: str) -> dict[str, Any]: log_entry["schedule_triggered_at"] = schedule_triggered_at state.setdefault("decision_log", []).append(log_entry) del state["decision_log"][:-50] - decision_persisted = _sync_decision_to_db(log_entry) candidate_evidence_valid = _decision_has_candidate_evidence(log_entry) - model_trades_persisted = _sync_trades_to_db(executed) - if executed and model_trades_persisted: - _sync_positions_to_db(state) record_equity(state) save_state(state) + model_persistence = _sync_committed_account_projections( + state, + trades=executed, + decisions=[log_entry], + ) + decision_persisted = model_persistence["decision_persisted"] + model_trades_persisted = model_persistence["trades_persisted"] model_executed = _accounted_trade_executions(executed) all_executed = [*position_exit_executed, *model_executed] if all_executed: @@ -12183,8 +12241,8 @@ def snapshot_closing_equity_once() -> dict[str, Any]: _refresh_position_bbi(state) rebuild_intraday_equity_curve(state, now=now) record_equity(state) - _sync_positions_to_db(state) save_state(state) + _sync_positions_to_db(state) today = now.strftime("%Y-%m-%d") closing_points = [ point @@ -12230,11 +12288,11 @@ def get_dashboard_payload() -> dict[str, Any]: refresh_today_sold_stocks(state) if not rebuild_intraday_equity_curve(state, now=now) and is_a_share_session_clock(now): record_equity(state) - _sync_positions_to_db(state) current_market_ctx = select_current_market_strategy_context(state, now) if current_market_ctx: state["market_decision_context"] = current_market_ctx save_state(state) + _sync_positions_to_db(state) payload = enrich_portfolio(state) payload["equity_history"] = load_account_history( diff --git a/scripts/docker-entrypoint.sh b/scripts/docker-entrypoint.sh index 12ba995..9b62c7e 100755 --- a/scripts/docker-entrypoint.sh +++ b/scripts/docker-entrypoint.sh @@ -77,9 +77,35 @@ export DASHBOARD_CRON_JOBS="$DASHBOARD_HOME/cron/jobs.json" export NIUONE_ROOT="$ROOT" umask 077 -mkdir -p \ +if ! mkdir -p \ "$DASHBOARD_HOME/cron/state" \ - "$DASHBOARD_HOME/logs" + "$DASHBOARD_HOME/cron/output" \ + "$DASHBOARD_HOME/logs"; then + echo "NiuOne runtime directories cannot be created by uid=$(id -u), gid=$(id -g): $DASHBOARD_HOME" >&2 + exit 73 +fi + +for runtime_dir in \ + "$DASHBOARD_HOME" \ + "$DASHBOARD_HOME/cron" \ + "$DASHBOARD_HOME/cron/state" \ + "$DASHBOARD_HOME/cron/output" \ + "$DASHBOARD_HOME/logs"; do + if [[ ! -w "$runtime_dir" ]]; then + echo "NiuOne runtime directory is not writable by uid=$(id -u), gid=$(id -g): $runtime_dir" >&2 + exit 73 + fi +done + +for runtime_file in \ + "$DASHBOARD_PORTFOLIO_STATE" \ + "$DASHBOARD_NIUNIU_DB" \ + "$DASHBOARD_PROMPT_STRATEGY_DB"; do + if [[ -e "$runtime_file" && ( ! -r "$runtime_file" || ! -w "$runtime_file" ) ]]; then + echo "NiuOne runtime file is not readable and writable by uid=$(id -u), gid=$(id -g): $runtime_file" >&2 + exit 73 + fi +done if [[ $# -eq 0 ]]; then set -- dashboard diff --git a/tests/test_container_deployment.py b/tests/test_container_deployment.py index 9b4ec1e..fddedcc 100644 --- a/tests/test_container_deployment.py +++ b/tests/test_container_deployment.py @@ -157,11 +157,83 @@ def test_entrypoint_keeps_container_paths_and_listener_invariants(self): self.assertIsNone(values["newsnow_process_base_url"]) self.assertEqual(values["bundled_newsnow_url"], "http://newsnow:4444/api/s") self.assertEqual(values["newsnow_endpoint"], "http://newsnow:4444/api/s") + self.assertTrue((data_dir / "runtime" / "cron" / "output").is_dir()) self.assertIn("DASHBOARD_RATE_LIMIT_ANON=241", values["persisted"]) self.assertNotIn("NIUONE_ROOT=", values["persisted"]) self.assertNotIn("DASHBOARD_LOG_DIR=", values["persisted"]) self.assertNotIn("DASHBOARD_B1_SCANNER=", values["persisted"]) + @unittest.skipIf( + os.name == "nt" or getattr(os, "geteuid", lambda: 0)() == 0, + "POSIX non-root permissions are required", + ) + def test_entrypoint_rejects_unwritable_account_runtime_directory(self): + with tempfile.TemporaryDirectory() as tmp: + data_dir = Path(tmp) / "data" + output_dir = data_dir / "runtime" / "cron" / "output" + output_dir.mkdir(parents=True) + output_dir.chmod(0o500) + env = os.environ.copy() + env.update({ + "NIUONE_CONTAINER_DATA_DIR": str(data_dir), + "DASHBOARD_ENV_FILE": str(data_dir / "missing.env"), + "PYTHON_BIN": sys.executable, + }) + try: + result = subprocess.run( + [ + "bash", + str(ROOT / "scripts" / "docker-entrypoint.sh"), + "true", + ], + cwd=ROOT, + env=env, + capture_output=True, + text=True, + ) + finally: + output_dir.chmod(0o700) + + self.assertEqual(result.returncode, 73) + self.assertIn("runtime directory is not writable", result.stderr) + + @unittest.skipIf( + os.name == "nt" or getattr(os, "geteuid", lambda: 0)() == 0, + "POSIX non-root permissions are required", + ) + def test_entrypoint_reports_uncreatable_runtime_directories(self): + with tempfile.TemporaryDirectory() as tmp: + data_dir = Path(tmp) / "data" + runtime_dir = data_dir / "runtime" + runtime_dir.mkdir(parents=True) + runtime_dir.chmod(0o500) + env = os.environ.copy() + env.update({ + "NIUONE_CONTAINER_DATA_DIR": str(data_dir), + "DASHBOARD_ENV_FILE": str(data_dir / "missing.env"), + "PYTHON_BIN": sys.executable, + }) + try: + result = subprocess.run( + [ + "bash", + str(ROOT / "scripts" / "docker-entrypoint.sh"), + "true", + ], + cwd=ROOT, + env=env, + capture_output=True, + text=True, + ) + finally: + runtime_dir.chmod(0o700) + + self.assertEqual(result.returncode, 73) + self.assertIn( + "runtime directories cannot be created", + result.stderr, + ) + def test_release_workflow_uses_tag_trigger_and_repository_credentials(self): path = ROOT / ".github" / "workflows" / "docker-publish.yml" text = path.read_text(encoding="utf-8") diff --git a/tests/test_sell_strategy_rules.py b/tests/test_sell_strategy_rules.py index bdfb4b5..f3c8282 100644 --- a/tests/test_sell_strategy_rules.py +++ b/tests/test_sell_strategy_rules.py @@ -3378,17 +3378,25 @@ def test_auto_exit_reports_durable_ledger_failure(self): "decision_log": [], } originals = { + "load_state": trader.load_state, + "save_state": trader.save_state, + "record_equity": trader.record_equity, "_sync_trades_to_db": trader._sync_trades_to_db, "_sync_decision_to_db": trader._sync_decision_to_db, } try: + trader.load_state = lambda: json.loads(json.dumps(state)) + trader.save_state = lambda _state: None + trader.record_equity = lambda _state: False trader._sync_trades_to_db = lambda _rows: False trader._sync_decision_to_db = lambda _row: True - executed = trader.check_auto_exits( - state, - datetime(2026, 6, 24, 9, 37), + _saved_state, executed, persistence = ( + trader._commit_refreshed_auto_exits( + state, + trader._auto_exit_refresh_baseline(state), + datetime(2026, 6, 24, 9, 37), + ) ) - persistence = trader._pop_auto_exit_persistence_status(state) finally: for name, value in originals.items(): setattr(trader, name, value) diff --git a/tests/test_trade_accounting.py b/tests/test_trade_accounting.py index 002f3ab..b379e98 100644 --- a/tests/test_trade_accounting.py +++ b/tests/test_trade_accounting.py @@ -129,6 +129,124 @@ def test_save_state_rejects_divergent_oversell_without_crediting_cash(self): self.assertFalse(trade_counts_for_account(rejected)) self.assertEqual(trader._trade_cash_delta(rejected), 0.0) + def test_save_state_commits_json_before_archiving_history(self): + trade = { + "time": "2026-08-17 10:00:00", + "action": "BUY", + "code": "600000", + "shares": 100, + "price": 10.0, + "amount": 1000.0, + "reason": "提交顺序测试", + } + observed_states = [] + + def archive_after_commit(_state): + observed_states.append( + json.loads(trader.STATE_FILE.read_text(encoding="utf-8")) + ) + return True + + trader._archive_account_history_before_compaction = archive_after_commit + trader.save_state(self._base_state( + cash=98_999.0, + trade_log=[trade], + )) + + self.assertEqual(len(observed_states), 1) + self.assertEqual(observed_states[0]["cash"], 98_999.0) + self.assertEqual(observed_states[0]["trade_log"], [trade]) + + def test_save_state_failure_before_commit_does_not_archive_history(self): + archive_calls = [] + original_writer = trader._write_state_file_atomically + + def fail_before_commit(_payload): + raise PermissionError("state file is not writable") + + try: + trader._write_state_file_atomically = fail_before_commit + trader._archive_account_history_before_compaction = ( + lambda _state: archive_calls.append(True) or True + ) + with self.assertRaises(PermissionError): + trader.save_state(self._base_state()) + finally: + trader._write_state_file_atomically = original_writer + + self.assertEqual(archive_calls, []) + + def test_compaction_failure_keeps_committed_full_history(self): + original_writer = trader._write_state_file_atomically + write_count = 0 + + def fail_compaction(payload): + nonlocal write_count + write_count += 1 + if write_count == 2: + raise PermissionError("compaction replace failed") + original_writer(payload) + + trader._archive_account_history_before_compaction = lambda _state: True + try: + trader._write_state_file_atomically = fail_compaction + trader.save_state(self._base_state( + trade_log=[ + { + "time": f"2026-08-17 10:{index // 60:02d}:{index % 60:02d}", + "action": "BUY", + "code": f"{index:06d}", + "shares": 100, + "price": 10.0, + "amount": 1000.0, + "reason": "压缩降级测试", + } + for index in range(trader.TRADE_LOG_LIMIT + 1) + ], + )) + finally: + trader._write_state_file_atomically = original_writer + + saved = json.loads(trader.STATE_FILE.read_text(encoding="utf-8")) + self.assertEqual(write_count, 2) + self.assertEqual(len(saved["trade_log"]), trader.TRADE_LOG_LIMIT + 1) + + def test_delayed_position_projection_reads_latest_canonical_state(self): + first_position = { + "600000": { + "code": "600000", + "qty": 100, + "avg_cost": 10.0, + } + } + latest_positions = { + **first_position, + "600001": { + "code": "600001", + "qty": 200, + "avg_cost": 8.0, + }, + } + stale_state = self._base_state(positions=first_position) + trader.save_state(self._base_state(positions=latest_positions)) + + captured = [] + original_db_module = sys.modules.get("niuniu_db") + sys.modules["niuniu_db"] = types.SimpleNamespace( + snapshot_positions=lambda positions: captured.append( + copy.deepcopy(positions) + ) + ) + try: + trader._sync_positions_to_db(stale_state) + finally: + if original_db_module is None: + sys.modules.pop("niuniu_db", None) + else: + sys.modules["niuniu_db"] = original_db_module + + self.assertEqual(captured, [latest_positions]) + def test_rejected_audit_marker_survives_same_trade_merge(self): trade = { "time": "2026-08-17 09:38:01", diff --git a/tests/test_trade_notification_hooks.py b/tests/test_trade_notification_hooks.py index 59d77be..4431913 100644 --- a/tests/test_trade_notification_hooks.py +++ b/tests/test_trade_notification_hooks.py @@ -79,21 +79,42 @@ def test_rejected_fill_is_not_returned_to_notification_dispatcher(self): def test_auto_exit_notifies_only_after_state_is_saved(self): events = [] executed = [sample_sell()] + + def check_auto_exits(state, _dt): + state.setdefault("trade_log", []).extend(executed) + state.setdefault("decision_log", []).append({ + "time": "2026-07-11 10:00:00", + "decision": {"summary": "测试自动离场"}, + "executed": executed, + }) + return executed + with patched( - load_state=lambda: {"positions": {}, "trade_log": [], "cash": 1000.0}, + load_state=lambda: { + "positions": {}, + "trade_log": [], + "decision_log": [], + "cash": 1000.0, + }, refresh_realtime_prices=lambda state: None, refresh_position_intraday=lambda state: None, _refresh_position_bbi=lambda state, dt=None: None, - check_auto_exits=lambda state, dt: executed, + check_auto_exits=check_auto_exits, record_equity=lambda state: None, save_state=lambda state: events.append("save"), + _sync_decision_to_db=lambda entry: events.append("decision") or True, + _sync_trades_to_db=lambda trades: events.append("trades") or True, + _sync_positions_to_db=lambda state: events.append("positions"), _notify_trade_executions_safely=lambda trades: events.append(("notify", trades)), enrich_portfolio=lambda state: {}, ): result = trader.run_auto_exits_once(datetime(2026, 7, 11, 10, 0)) self.assertEqual(result["executed"], executed) - self.assertEqual(events, ["save", ("notify", executed)]) + self.assertEqual( + events, + ["save", "decision", "trades", "positions", ("notify", executed)], + ) def test_auto_exit_with_no_fill_does_not_notify(self): events = [] @@ -136,9 +157,9 @@ def test_deferred_fill_notifies_once_after_state_is_saved(self): refine_overlimit_buy_actions=lambda *args, **kwargs: {}, execute_actions=lambda *args, **kwargs: executed, enrich_portfolio=lambda value: {}, - _sync_decision_to_db=lambda entry: None, - _sync_trades_to_db=lambda trades: None, - _sync_positions_to_db=lambda value: None, + _sync_decision_to_db=lambda entry: events.append("decision") or True, + _sync_trades_to_db=lambda trades: events.append("trades") or True, + _sync_positions_to_db=lambda value: events.append("positions"), record_equity=lambda value: None, save_state=lambda value: events.append("save"), _notify_trade_executions_safely=lambda trades: events.append(("notify", trades)), @@ -146,7 +167,10 @@ def test_deferred_fill_notifies_once_after_state_is_saved(self): result = trader.execute_due_pending_decisions(datetime(2026, 7, 11, 13, 0)) self.assertEqual(result["executed"], executed) - self.assertEqual(events, ["save", ("notify", executed)]) + self.assertEqual( + events, + ["save", "decision", "trades", "positions", ("notify", executed)], + ) def test_model_fill_notifies_once_after_state_is_saved(self): events = [] @@ -179,9 +203,9 @@ def test_model_fill_notifies_once_after_state_is_saved(self): call_model_decision=lambda *args, **kwargs: decision, refine_overlimit_buy_actions=lambda *args, **kwargs: {}, execute_actions=lambda *args, **kwargs: executed, - _sync_decision_to_db=lambda entry: None, - _sync_trades_to_db=lambda trades: None, - _sync_positions_to_db=lambda value: None, + _sync_decision_to_db=lambda entry: events.append("decision") or True, + _sync_trades_to_db=lambda trades: events.append("trades") or True, + _sync_positions_to_db=lambda value: events.append("positions"), record_equity=lambda value: None, save_state=lambda value: events.append("save"), _notify_trade_executions_safely=lambda trades: events.append(("notify", trades)), @@ -189,7 +213,169 @@ def test_model_fill_notifies_once_after_state_is_saved(self): result = trader.run_decision_after_b1({"generated_at": "2026-07-11 10:00:00"}, force=True) self.assertEqual(result["executed"], executed) - self.assertEqual(events, ["save", ("notify", executed)]) + self.assertEqual( + events, + ["save", "decision", "trades", "positions", ("notify", executed)], + ) + + def test_model_fill_state_failure_does_not_project_or_notify(self): + events = [] + executed = [sample_sell()] + state = { + "cash": 1000.0, + "positions": { + "600000": {"qty": 100, "avg_cost": 10.0, "last_price": 10.5} + }, + "trade_log": [], + "decision_log": [], + "equity_history": [], + } + + def fail_save(_state): + events.append("save") + raise PermissionError("state file is not writable") + + with patched( + load_state=lambda: state, + market_strategy_context_for_b1=lambda payload: { + "tone_label": "中性", + "max_open_positions": 6, + "max_new_buys_per_decision": 2, + "allow_new_buys": True, + }, + compact_market_strategy_context=lambda value: value, + run_position_exit_checks_before_decision=lambda state, dt=None: [], + check_daily_loss_budget=lambda value: (False, 0.0), + get_adaptive_params=lambda: {}, + is_a_share_execution_time=lambda now=None: (True, "连续竞价交易时段"), + check_market_environment=lambda: {"bullish": True}, + check_market_sentiment=lambda: {"sentiment": "neutral", "detail": ""}, + enrich_portfolio=lambda value: {}, + call_model_decision=lambda *args, **kwargs: { + "summary": "测试决策", + "actions": [{"action": "SELL", "code": "600000", "shares": 100}], + }, + refine_overlimit_buy_actions=lambda *args, **kwargs: {}, + execute_actions=lambda *args, **kwargs: executed, + _sync_decision_to_db=lambda entry: events.append("decision") or True, + _sync_trades_to_db=lambda trades: events.append("trades") or True, + _sync_positions_to_db=lambda value: events.append("positions"), + record_equity=lambda value: None, + save_state=fail_save, + _notify_trade_executions_safely=lambda trades: events.append("notify"), + ): + with self.assertRaises(PermissionError): + trader.run_decision_after_b1( + {"generated_at": "2026-07-11 10:00:00"}, + force=True, + ) + + self.assertEqual(events, ["save"]) + + def test_auto_exit_state_failure_does_not_project_or_notify(self): + events = [] + executed = [sample_sell()] + + def check_auto_exits(state, _dt): + state.setdefault("trade_log", []).extend(executed) + state.setdefault("decision_log", []).append({ + "time": "2026-07-11 10:00:00", + "decision": {"summary": "测试自动离场"}, + "executed": executed, + }) + return executed + + def fail_save(_state): + events.append("save") + raise PermissionError("state file is not writable") + + with patched( + load_state=lambda: { + "positions": {}, + "trade_log": [], + "decision_log": [], + "cash": 1000.0, + }, + refresh_realtime_prices=lambda state: None, + refresh_position_intraday=lambda state: None, + _refresh_position_bbi=lambda state, dt=None: None, + check_auto_exits=check_auto_exits, + record_equity=lambda state: None, + save_state=fail_save, + _sync_decision_to_db=lambda entry: events.append("decision") or True, + _sync_trades_to_db=lambda trades: events.append("trades") or True, + _sync_positions_to_db=lambda state: events.append("positions"), + _notify_trade_executions_safely=lambda trades: events.append("notify"), + enrich_portfolio=lambda state: {}, + ): + with self.assertRaises(PermissionError): + trader.run_auto_exits_once(datetime(2026, 7, 11, 10, 0)) + + self.assertEqual(events, ["save"]) + + def test_deferred_fill_state_failure_does_not_project_or_notify(self): + events = [] + state = { + "cash": 1000.0, + "positions": {}, + "trade_log": [], + "decision_log": [], + "pending_decisions": [{ + "id": "pending-1", + "status": "pending", + "due_at": "", + "decision": {"summary": "延迟测试", "actions": []}, + "candidates": [], + "schedule_slot": "2026-07-11 09:25", + }], + } + + def fail_save(_state): + events.append("save") + raise PermissionError("state file is not writable") + + with patched( + is_a_share_execution_time=lambda now=None: (True, "连续竞价交易时段"), + load_state=lambda: state, + current_market_strategy_context=lambda: {}, + refine_overlimit_buy_actions=lambda *args, **kwargs: {}, + execute_actions=lambda *args, **kwargs: [sample_sell()], + enrich_portfolio=lambda value: {}, + _sync_decision_to_db=lambda entry: events.append("decision") or True, + _sync_trades_to_db=lambda trades: events.append("trades") or True, + _sync_positions_to_db=lambda value: events.append("positions"), + record_equity=lambda value: None, + save_state=fail_save, + _notify_trade_executions_safely=lambda trades: events.append("notify"), + ): + with self.assertRaises(PermissionError): + trader.execute_due_pending_decisions( + datetime(2026, 7, 11, 13, 0) + ) + + self.assertEqual(events, ["save"]) + + def test_decision_log_state_failure_does_not_project(self): + events = [] + + def fail_save(_state): + events.append("save") + raise PermissionError("state file is not writable") + + with patched( + load_state=lambda: {"decision_log": []}, + save_state=fail_save, + _sync_decision_to_db=lambda entry: events.append("decision") or True, + ): + with self.assertRaises(PermissionError): + trader.record_decision_log_entry({ + "time": "2026-07-11 10:00:00", + "b1_generated_at": "", + "decision": {"summary": "只记录决策"}, + "executed": [], + }) + + self.assertEqual(events, ["save"]) def test_position_exit_check_runs_before_model_even_without_candidates(self): events = []