# -*- coding: utf-8 -*- """ 宏观择时 · 编排 (取数 → 计算 → 落表 → 决策 → 分流 → 留痕) ============================================================ 设计: MACRO_TIMING_PLAN.md (V3)。参数依据: 标定报告 MACRO_CALIB_2026-08-18.md。 铁律落点: * 唯一出口 —— 仓位变化只经 command_service.issue() 产生 (署名 macro), 加上一个 只拦增持的「宏观闸」; 不直接写方案/指令/持仓账本, 不碰执行器。 * 先记账后动作 —— 信号先落 pms_macro_signal, 再决策下命令; 命令本身走 issue 的 先落表路径。 * 故障即守成 —— 取数/计算/读组合任一失败: 不下命令、不落闸 (闸的安全方向是不额外拦, 增持自有规则闸把关), 页面亮黄条。 * 命令至上 —— 与用户在途命令冲突 (issue 返回 CONFLICT) 即放弃并留痕, 永不 force。 宏观闸的状态载体是运行参数 PMS_MACRO_GATE_STATE (JSON), 由本模块每次扫描时写入, proposal_service 与 strategy_runner 经 gate_state() 只读 —— 不用表、不建消费组, 与 PMS_STRATEGY_BUYPAUSE / PMS_CUSHION_NEG_STREAK 用独立运行参数承载状态是同一个手法。 调度位在 scheduler.macro_scan (交易日 09:35); 手动入口 POST /api/ops/macro-scan。 """ from __future__ import annotations import json import logging import threading import time from datetime import datetime from app.core import macro_rules as mr from app.core import tradedays as td from app.repo import macro_repo, pms_repo from app.services import param_store logger = logging.getLogger("pms.macro") GATE_KEY = "PMS_MACRO_GATE_STATE" GATE_FRESH_DAYS = 7 # 闸状态超过 7 个自然日没刷新 → 视为失效 (调度断了别让闸卡死) # 信号注册表: 接新宏观信号只加这里 + 各自参数 (命名空间 PMS_MACRO__*), # 决策/执行/闸/页面零改动。fetch 返回原始序列, compute 是纯函数。 # 用 lambda 包一层做**晚绑定**: 调用时才解析 macro_repo 的模块属性 —— 单测打桩 # (替换模块函数) 才打得到; 直接存函数引用会把打桩绕过去, 测试全走真库。 SIGNALS = { "hedge_fx": {"label": "股汇对冲指数", "fetch": lambda: macro_repo.fetch_hedge_inputs(), "compute": lambda inputs, p, today: mr.compute_hedge_index( inputs, p, today)}, } A_NONE, A_ADV_RED, A_ADV_INC = "NONE", "ADVICE_REDUCE", "ADVICE_INCREASE" A_CMD, A_BLOCKED = "CMD_ISSUED", "BLOCKED" CMD_OF = {mr.ACT_REDUCE: "REDUCE_EXPOSURE", mr.ACT_INCREASE: "INCREASE_EXPOSURE"} ADV_OF = {mr.ACT_REDUCE: A_ADV_RED, mr.ACT_INCREASE: A_ADV_INC} _gate_cache = {"at": 0.0, "data": None} _gate_lock = threading.Lock() # ================================================================ 参数 def _params() -> dict: """一次取齐全部宏观参数 (键名小写化, 供 macro_rules 纯函数用)。""" g = param_store return { "enabled": g.get_bool("PMS_MACRO_ENABLED", False), "autonomy": str(g.get("PMS_MACRO_AUTONOMY", "propose_only") or "propose_only"), "signals": g.get_list("PMS_MACRO_SIGNALS", ["hedge_fx"]), "stock_gate": g.get_bool("PMS_MACRO_STOCK_GATE", True), "hot_th": g.get_float("PMS_MACRO_HOT_TH", 25.0), "cold_th": g.get_float("PMS_MACRO_COLD_TH", -20.0), "exit_band": g.get_float("PMS_MACRO_EXIT_BAND", 15.0), "confirm_days": g.get_int("PMS_MACRO_CONFIRM_DAYS", 1), "trigger_hot": str(g.get("PMS_MACRO_TRIGGER_HOT", mr.TRIG_EXIT) or mr.TRIG_EXIT), "trigger_cold": str(g.get("PMS_MACRO_TRIGGER_COLD", mr.TRIG_ENTER) or mr.TRIG_ENTER), "log_s0": g.get_float("PMS_MACRO_LOG_S0", 0.049), "log_k": g.get_float("PMS_MACRO_LOG_K", 1.6), "shift_max": g.get_float("PMS_MACRO_SHIFT_MAX", 0.20), "step_min": g.get_float("PMS_MACRO_STEP_MIN", 0.02), "cooldown_tdays": g.get_int("PMS_MACRO_COOLDOWN_TDAYS", 3), "min_pct": g.get_float("PMS_MACRO_MIN_PCT", 0.20), "max_pct": g.get_float("PMS_MACRO_MAX_PCT", 0.0), "window_tdays": g.get_int("PMS_MACRO_WINDOW_TDAYS", 3), "respect_brake": g.get_bool("PMS_MACRO_RESPECT_BRAKE", True), "stale_tdays": g.get_int("PMS_MACRO_STALE_TDAYS", 3), "ret_win": g.get_int("PMS_MACRO_RET_WIN", 20), "z_win": g.get_int("PMS_MACRO_Z_WIN", 40), "beta": g.get_float("PMS_MACRO_SHIBOR_BETA", 0.02), } def _enabled_signals(p: dict) -> list: return [k for k in (p.get("signals") or []) if k in SIGNALS] # ================================================================ 扫描主流程 def scan(*, dry_run: bool = False, now=None) -> dict: """每交易日一次的宏观扫描 (调度 09:35; 手动端点随时可跑, dry_run 只算不落表不下达)。 非交易日经手动端点跑到这里不拦 (调度位自有交易日守卫), 只是行日期按自然日记。 """ now = now or datetime.now() today = td.ymd(now) p = _params() out = {"ok": True, "enabled": p["enabled"], "autonomy": p["autonomy"], "dry_run": dry_run, "signals": [], "actions": [], "skipped": [], "errors": []} if not p["enabled"]: out["skipped"].append("宏观择时总开关关闭 (PMS_MACRO_ENABLED=False)") return out if param_store.get_bool("PMS_GLOBAL_EXEC_HALT", False): # 调度守卫已挡休假; 这里再兜手动端点这条路 out["skipped"].append("全局暂停执行 (休假模式), 宏观扫描跳过") return out hot_now = False for key in _enabled_signals(p): try: r = _scan_one(key, SIGNALS[key], p, today, now, dry_run, out) except Exception as e: # noqa: BLE001 —— 单信号异常不拖垮整跳 (守成) logger.exception("[宏观] 信号 %s 扫描失败", key) out["errors"].append(f"{key}: {type(e).__name__}: {e}") continue out["signals"].append(r) if r.get("zone") == mr.Z_HOT: hot_now = True # ---- 宏观闸状态落参数 (增持闸; off 档只看不动, 闸也不生效) ---- gate_active = bool(p["stock_gate"] and p["autonomy"] != "off" and hot_now) out["gate"] = {"active": gate_active, "ymd": today} if not dry_run: w = param_store.set_param(GATE_KEY, json.dumps( {"active": gate_active, "ymd": today, "why": "股汇对冲指数偏热确认" if gate_active else ""}, ensure_ascii=False), "system") or {} if not w.get("ok"): # 闸状态没写上要吭声: 状态陈旧超 7 天 gate_state 会自动失效 (失效方向是不拦) logger.error("[宏观] 闸状态写入失败: %s —— 闸维持上一次的旧状态", w.get("error")) out["errors"].append(f"闸状态写入失败: {w.get('error')}") with _gate_lock: _gate_cache["at"] = 0.0 # 失效缓存, 下一读取新状态 out["ok"] = not out["errors"] return out def _scan_one(key: str, spec: dict, p: dict, today: int, now, dry_run: bool, out: dict) -> dict: """单个信号: 取数 → 计算 → 区域与周期 → 落表 → 决策 → 分流。返回页面摘要。""" label = spec.get("label") or key # ---- 取数与计算 (失败 → UNAVAILABLE 行, 不动作) ---- try: inputs = spec["fetch"]() comp = spec["compute"](inputs, p, today) except Exception as e: # noqa: BLE001 logger.error("[宏观] %s 取数/计算失败: %s", key, e) if not dry_run: macro_repo.upsert_signal(signal_key=key, trade_date=today, value=None, zone=mr.Z_UNAVAILABLE, detail={}, note=f"取数失败: {type(e).__name__}: {e}") return {"key": key, "label": label, "zone": mr.Z_UNAVAILABLE, "value": None, "note": f"取数失败: {e}"} # ---- 昨日状态与同日重扫合并 ---- try: history = macro_repo.recent_signals(key, limit=15) except Exception as e: # noqa: BLE001 —— 历史读不到按冷启动处理 (保守: 无周期无冷却豁免) logger.warning("[宏观] %s 历史读取失败 (按冷启动): %s", key, e) history = [] prev_row = next((r for r in history if int(r["trade_date"]) < today), None) today_row = next((r for r in history if int(r["trade_date"]) == today), None) prev_zone = (prev_row or {}).get("zone") or mr.Z_NEUTRAL prev_cycle = ((prev_row or {}).get("detail") or {}).get("cycle") if not comp["ok"]: if not dry_run: macro_repo.upsert_signal(signal_key=key, trade_date=today, value=None, zone=mr.Z_UNAVAILABLE, detail=comp.get("detail"), note=comp.get("reason") or "数据守卫未过") return {"key": key, "label": label, "zone": mr.Z_UNAVAILABLE, "value": None, "note": comp.get("reason")} value = comp["value"] zone = mr.zone_next(value, prev_zone, p) cycle, exit_event = mr.carry_cycle(prev_zone, prev_cycle, zone, value, p, today) t_detail = (today_row or {}).get("detail") or {} t_cycle = t_detail.get("cycle") or {} if cycle and t_cycle.get("zone") == cycle.get("zone"): # 同日重扫: 当日已执行量与峰值只增不减 (免得重扫把已下的命令"忘掉"再下一次) cycle["done_shift"] = max(cycle["done_shift"], float(t_cycle.get("done_shift") or 0)) cycle["e_peak"] = max(cycle["e_peak"], float(t_cycle.get("e_peak") or 0)) exit_acted = bool(t_detail.get("exit_acted")) e_now = mr.depth_of(value, zone, p) detail = {**(comp.get("detail") or {}), "cycle": cycle, "e_now": round(e_now, 2)} if exit_acted: detail["exit_acted"] = True # 当日已下过命令的行不回退动作字段与说明 (先记账后动作: 命令已在命令表, 行只增不改口) action_field = A_CMD if (today_row or {}).get("action") == A_CMD else A_NONE ref_id = (today_row or {}).get("ref_id") or "" note = ((today_row or {}).get("note") if action_field == A_CMD else None) \ or f"区域 {zone}, 指数 {value:+.1f}" if not dry_run: macro_repo.upsert_signal(signal_key=key, trade_date=today, value=value, zone=zone, detail=detail, action=action_field, ref_id=ref_id, note=note) # ---- 决策 ---- flags, flag_err = _flags(p, key, history, today) if flag_err: out["errors"].append(f"{key}: {flag_err}") return {"key": key, "label": label, "zone": zone, "value": value, "note": f"{note}; 组合状态读取失败, 本轮不决策"} d = mr.decide(zone=zone, cycle=cycle, exit_event=exit_event, exit_acted=exit_acted, e_now=e_now, flags=flags, params=p) summary = {"key": key, "label": label, "zone": zone, "value": value, "e_now": round(e_now, 2), "streak": (cycle or {}).get("streak"), "note": note} if action_field == A_CMD: summary["note"] = note + "; 当日已下过宏观命令, 本轮不重复" return summary # ---- 想动但被拦: 留痕 (与当日已记录的同因不重复写) ---- if d.get("blocked"): why = "; ".join(d["blocked"])[:400] if not dry_run: _upsert_action(key, today, value, zone, detail, A_BLOCKED, "", why) _ledger_once(today_row, A_BLOCKED, why, key, value, detail, verdict="REJECT") summary["note"] = why out["skipped"].append({"signal": key, "why": why}) return summary if not d.get("action"): return summary # ---- 触发: 按档位分流 ---- direction, step = d["action"], float(d["pct"]) why = f"{label} {value:+.1f} · {d.get('why') or ''}".strip() if p["autonomy"] == "off": summary["note"] = f"只看不动档: 想{_cn(direction)} {step:.1%} ({why})" out["skipped"].append({"signal": key, "why": summary["note"]}) return summary if dry_run: out["actions"].append({"signal": key, "action": direction, "pct": step, "route": "dry_run", "why": why}) summary["note"] = f"试算: 将{_cn(direction)} {step:.1%}" return summary if p["autonomy"] == "propose_only": adv = ADV_OF[direction] detail["advice"] = {"direction": direction, "pct": round(step, 4), "why": why, "at": str(now)[:19]} adv_note = f"建议{_cn(direction)} {step:.1%} (待页面采纳, 当日有效)" _upsert_action(key, today, value, zone, detail, adv, "", adv_note) _ledger_once(today_row, adv, adv_note, key, value, detail, verdict="NOTE") out["actions"].append({"signal": key, "action": direction, "pct": step, "route": "advice", "why": why}) summary["note"] = adv_note return summary # full: 直接下命令 (issued_by=macro; 冲突即弃, 永不 force) r = _issue(direction, step, p, why, issued_by="macro") if r.get("ok"): cid = r.get("command_id") or "" _mark_acted(detail, direction, step, p) _upsert_action(key, today, value, zone, detail, A_CMD, cid, f"已下{_cn(direction)}命令 {step:.1%} → {cid}") _ledger_once(None, A_CMD, why, key, value, detail, verdict="PASS", ref_id=cid) out["actions"].append({"signal": key, "action": direction, "pct": step, "route": "command", "command_id": cid, "why": why}) summary["note"] = f"已下{_cn(direction)}命令 {cid}" else: errs = "; ".join(str(x) for x in (r.get("errors") or [])) or "下达失败" blocked_note = f"想{_cn(direction)} {step:.1%} 但命令未成: {errs}"[:400] _upsert_action(key, today, value, zone, detail, A_BLOCKED, "", blocked_note) _ledger_once(today_row, A_BLOCKED, blocked_note, key, value, detail, verdict="REJECT") out["skipped"].append({"signal": key, "why": blocked_note}) summary["note"] = blocked_note return summary def _cn(direction: str) -> str: return "降仓" if direction == mr.ACT_REDUCE else "升仓" def _issue(direction: str, step: float, p: dict, why: str, *, issued_by: str) -> dict: from app.services import command_service return command_service.issue( CMD_OF[direction], {"pct": round(step, 4), "window_tdays": p["window_tdays"]}, issued_by=issued_by, note=f"宏观择时: {why}"[:280]) or {} def _mark_acted(detail: dict, direction: str, step: float, p: dict): """动作落地后推进周期状态: 进区模式累计已执行量; 回落模式记当日已动作。""" mode = p["trigger_hot"] if direction == mr.ACT_REDUCE else p["trigger_cold"] cyc = detail.get("cycle") if mode == mr.TRIG_ENTER and cyc: cyc["done_shift"] = round(float(cyc.get("done_shift") or 0) + step, 4) else: detail["exit_acted"] = True def _upsert_action(key, today, value, zone, detail, action, ref_id, note): try: macro_repo.upsert_signal(signal_key=key, trade_date=today, value=value, zone=zone, detail=detail, action=action, ref_id=ref_id, note=note) except Exception as e: # noqa: BLE001 —— 行没写上要吭声, 命令/建议本体已各自落表 logger.error("[宏观] %s 动作回写信号行失败: %s", key, e) def _ledger_once(today_row, action_field, note, key, value, detail, *, verdict, ref_id=None): """评审账本留痕; 同日同动作同说明不重复写 (手动重扫防刷屏)。""" if today_row and today_row.get("action") == action_field \ and (today_row.get("note") or "") == (note or "")[:500]: return try: pms_repo.insert_ledger( ts_code="-", action="MACRO", arbiter="rule", verdict=verdict, price_at=0, hard_numbers={"signal": key, "value": value, "e_now": (detail or {}).get("e_now"), "cycle": (detail or {}).get("cycle")}, ref_id=ref_id, reason=(note or "")[:500]) except Exception as e: # noqa: BLE001 logger.warning("[宏观] 评审账本留痕失败: %s", e) # ================================================================ 让路状态 def _flags(p: dict, key: str, history: list, today: int) -> tuple: """备齐 decide 所需的让路旗标。组合快照读不到 → 返回错误, 本轮不决策 (守成)。""" try: from app.services import portfolio view = portfolio.positions_view() pf = float(view["totals"].get("portfolio_pct") or 0.0) cap = float(view["params"].get("portfolio_cap") or 1.0) except Exception as e: # noqa: BLE001 return {}, f"读组合快照失败: {type(e).__name__}: {e}" inflight, last_red, last_inc = _action_history(history) return { "portfolio_pct": pf, "portfolio_cap": cap, "buy_halt": param_store.get_bool("PMS_GLOBAL_BUY_HALT", False), "brake_active": today < param_store.get_int("PMS_BRAKE_UNTIL", 0), "inflight": inflight, "cooldown_left_reduce": _cooldown_left(history, last_red, today, p["cooldown_tdays"]), "cooldown_left_increase": _cooldown_left(history, last_inc, today, p["cooldown_tdays"]), }, None def _action_history(history: list) -> tuple: """从信号历史行回查: 是否有署名 macro 的命令在途 + 两个方向最近一次生效动作日。 「生效」= 所发命令没有被立刻取消 (零方案命令不占冷却, 次日可重试)。 只查最近几条 CMD_ISSUED 行, 逐条看命令状态 —— 不新增 repo 查询面。 """ from app.core import command_spec as cs inflight, last_red, last_inc = False, None, None looked = 0 for r in history: if r.get("action") != A_CMD or not r.get("ref_id"): continue looked += 1 if looked > 5: break try: cmd = pms_repo.get_command(r["ref_id"]) except Exception: # noqa: BLE001 —— 查不到按保守方向: 当作在途, 今天不动 logger.warning("[宏观] 命令 %s 状态查不到, 按在途处理 (保守)", r.get("ref_id")) inflight = True continue if not cmd: continue if cmd.get("status") in cs.ACTIVE_TASK_STATES: inflight = True if cmd.get("status") != cs.ST_CANCELLED: d = ((r.get("detail") or {}).get("advice") or {}).get("direction") \ or _dir_of_cmd(cmd.get("cmd_type")) ymd = int(r["trade_date"]) if d == mr.ACT_REDUCE: last_red = max(last_red or 0, ymd) elif d == mr.ACT_INCREASE: last_inc = max(last_inc or 0, ymd) return inflight, last_red, last_inc def _dir_of_cmd(cmd_type) -> str: return {v: k for k, v in CMD_OF.items()}.get(str(cmd_type or ""), "") def _cooldown_left(history: list, last_ymd, today: int, cooldown: int) -> int: """冷却剩余交易日。交易日距离按信号历史行数 (每交易日一行); 缺行只会把冷却算长, 方向保守。last_ymd 为空 = 从未动过, 无冷却。""" if not last_ymd: return 0 dist = len({int(r["trade_date"]) for r in history if last_ymd < int(r["trade_date"]) <= today}) if not any(int(r["trade_date"]) == today for r in history): dist += 1 # 今天的行还没写 (本跳首扫), 今天也算一天 return max(0, int(cooldown) - dist) # ================================================================ 宏观闸 (只读) def gate_state() -> dict: """个股宏观闸状态 (proposal_service / strategy_runner 每跳来读, 5 秒缓存)。 任何一步失败一律返回不生效 —— 闸的安全方向是「不额外拦」: 买入本身另有规则闸把关, 不能让宏观层故障把整个自主引擎摁死。状态超过 GATE_FRESH_DAYS 个自然日没刷新 (调度断了) 同样失效。 """ now = time.time() with _gate_lock: if _gate_cache["data"] is not None and now - _gate_cache["at"] < 5.0: return _gate_cache["data"] out = {"active": False} try: if param_store.get_bool("PMS_MACRO_ENABLED", False) \ and param_store.get_bool("PMS_MACRO_STOCK_GATE", True) \ and str(param_store.get("PMS_MACRO_AUTONOMY", "propose_only")) != "off": raw = param_store.get(GATE_KEY, "") or "" st = json.loads(raw) if raw else {} if st.get("active") and st.get("ymd"): if mr.ymd_gap_days(td.ymd(), int(st["ymd"])) <= GATE_FRESH_DAYS: out = {"active": True, "ymd": int(st["ymd"]), "why": st.get("why") or "宏观偏热确认"} else: out = {"active": False, "stale": True} except Exception as e: # noqa: BLE001 logger.warning("[宏观] 闸状态读取失败 (按不生效): %s", e) out = {"active": False, "error": str(e)} with _gate_lock: _gate_cache.update({"at": now, "data": out}) return out # ================================================================ 页面 def status() -> dict: """面板快照: 各信号的当前值/区域/近 20 日序列/当日建议/最近命令/冷却, 加闸状态。""" p = _params() today = td.ymd() out = {"ok": True, "enabled": p["enabled"], "autonomy": p["autonomy"], "gate": gate_state(), "params": { "hot_th": p["hot_th"], "cold_th": p["cold_th"], "exit_band": p["exit_band"], "cooldown_tdays": p["cooldown_tdays"], "trigger_hot": p["trigger_hot"], "trigger_cold": p["trigger_cold"]}, "signals": []} for key in _enabled_signals(p): item = {"key": key, "label": SIGNALS[key].get("label") or key} try: rows = macro_repo.recent_signals(key, limit=22) except Exception as e: # noqa: BLE001 item["error"] = f"{type(e).__name__}: {e}" out["signals"].append(item) continue latest = rows[0] if rows else None item["history"] = [{"ymd": int(r["trade_date"]), "value": (None if r.get("value") is None else float(r["value"])), "zone": r.get("zone")} for r in reversed(rows)] if latest: det = latest.get("detail") or {} item.update({ "ymd": int(latest["trade_date"]), "value": None if latest.get("value") is None else float(latest["value"]), "zone": latest.get("zone"), "note": latest.get("note"), "action": latest.get("action"), "ref_id": latest.get("ref_id"), "e_now": det.get("e_now"), "cycle": det.get("cycle"), "advice": (det.get("advice") if int(latest["trade_date"]) == today and str(latest.get("action") or "").startswith("ADVICE") else None), "data_ages": det.get("ages_days"), }) inflight, last_red, last_inc = _action_history(rows) item["inflight"] = inflight item["cooldown_left"] = { "reduce": _cooldown_left(rows, last_red, today, p["cooldown_tdays"]), "increase": _cooldown_left(rows, last_inc, today, p["cooldown_tdays"])} last_cmd = next((r for r in rows if r.get("action") == A_CMD and r.get("ref_id")), None) if last_cmd: brief = {"command_id": last_cmd.get("ref_id"), "ymd": int(last_cmd["trade_date"])} try: cmd = pms_repo.get_command(last_cmd["ref_id"]) if cmd: prog = cmd.get("progress") or {} brief.update({"status": cmd.get("status"), "cmd_type": cmd.get("cmd_type"), "done_amount": prog.get("done_amount"), "target_amount": prog.get("target_amount")}) except Exception: # noqa: BLE001 pass item["last_command"] = brief out["signals"].append(item) return out # ================================================================ 建议采纳 def adopt(signal_key: str = "hedge_fx") -> dict: """页面一键采纳当日建议 → 以用户名义下命令 (用户点头就是用户意志)。幂等: 只能采纳一次。""" p = _params() today = td.ymd() if not p["enabled"]: return {"ok": False, "error": "宏观择时总开关关闭"} try: row = macro_repo.get_signal(signal_key, today) except Exception as e: # noqa: BLE001 return {"ok": False, "error": f"读当日信号失败: {type(e).__name__}: {e}"} if not row: return {"ok": False, "error": "今天还没有信号记录 (先扫描)"} if row.get("action") == A_CMD: return {"ok": False, "error": f"当日已下过命令 {row.get('ref_id')}, 不重复采纳"} adv = (row.get("detail") or {}).get("advice") or {} if not str(row.get("action") or "").startswith("ADVICE") or not adv.get("direction"): return {"ok": False, "error": "当日没有待采纳的建议 (建议隔日作废, 次日重新评估)"} if param_store.get_bool("PMS_GLOBAL_EXEC_HALT", False): return {"ok": False, "error": "全局暂停执行 (休假模式) 生效中"} direction, step = adv["direction"], float(adv.get("pct") or 0) if direction == mr.ACT_INCREASE and param_store.get_bool("PMS_GLOBAL_BUY_HALT", False): return {"ok": False, "error": "全局暂停买入生效中, 升仓建议不可采纳"} if step <= 0: return {"ok": False, "error": f"建议幅度非法 ({step})"} r = _issue(direction, step, p, adv.get("why") or "宏观建议", issued_by="user") if not r.get("ok"): return {"ok": False, "error": "; ".join(str(x) for x in (r.get("errors") or [])) or "命令下达失败", "conflicts": r.get("conflicts")} cid = r.get("command_id") or "" detail = dict(row.get("detail") or {}) detail["advice"] = {**adv, "adopted": True, "command_id": cid} _mark_acted(detail, direction, step, p) _upsert_action(signal_key, today, row.get("value"), row.get("zone"), detail, A_CMD, cid, f"建议已采纳 → {cid}") _ledger_once(None, A_CMD, f"宏观建议采纳: {_cn(direction)} {step:.1%}", signal_key, row.get("value"), detail, verdict="PASS", ref_id=cid) return {"ok": True, "command_id": cid, "direction": direction, "pct": step}