diff --git a/mofin_db.py b/mofin_db.py index f653a13c..a949831d 100644 --- a/mofin_db.py +++ b/mofin_db.py @@ -1136,17 +1136,29 @@ def sync_recommend_tag(conn, code: str, timing_signal: str): def enqueue_recommend(conn, code: str): - """新推荐入摘要队列(防逐只轰炸)。batch_reassess 跑完后 flush_rec_digest 统一发一条。""" + """新推荐入摘要队列(防逐只轰炸)。batch_reassess 跑完后 flush_rec_digest 统一发一条。 + 校验:必须 tag=current_recommend 且信号为动作级,否则拒绝入队。""" try: import json as _j from datetime import datetime as _dt row = conn.execute( - "SELECT name, timing_signal, entry_low, entry_high, stop_loss, take_profit, " - "rr_ratio, position_advice FROM holding_strategies WHERE code=? AND status='active'", + "SELECT name, timing_signal, tag, entry_low, entry_high, stop_loss, take_profit, " + "rr_ratio, position_advice, full_analysis FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone() if not row: return - name, sig, el, eh, sl, tp, rr, pos = row + name, sig, tag, el, eh, sl, tp, rr, pos, fa = row + if tag != 'current_recommend' or sig not in ("买入", "可买入", "可加仓", "卖出", "止盈"): + print(f" [REC] {code} 非有效推荐(tag={tag},sig={sig}),不入队", flush=True) + return False + # 提取【最终新策略】段作为推荐依据摘要 + fa_text = fa or "" + strat = "" + for marker in ("【最终新策略】", "【综合结论】"): + idx = fa_text.find(marker) + if idx >= 0: + strat = fa_text[idx:idx + 450] + break qf = '/home/hmo/MoFin/gateway/logs/rec_digest_queue.jsonl' import os as _os _os.makedirs(_os.path.dirname(qf), exist_ok=True) @@ -1154,15 +1166,20 @@ def enqueue_recommend(conn, code: str): f.write(_j.dumps({"code": code, "name": name, "signal": sig, "entry_low": el, "entry_high": eh, "stop_loss": sl, "take_profit": tp, "rr": rr, "position": pos, + "strategy_excerpt": strat, + "full_analysis": fa_text[:2500], "ts": _dt.now().isoformat()}, ensure_ascii=False) + "\n") print(f" [REC] {code} 已入推荐摘要队列", flush=True) + return True except Exception as e: print(f" [REC] {code} 入队失败: {e}", flush=True) + return False def flush_rec_digest(max_items=5): - """把队列里的新推荐聚成一条 XMPP 摘要(RR 降序,最多 max_items 只)。""" - import json as _j, os as _os + """把队列里的新推荐聚成一条 XMPP 摘要(RR 降序,最多 max_items 只)。 + 头部 1-2 只附带策略依据摘要+按现金的操盘建议。""" + import json as _j, os as _os, sqlite3 as _sq qf = '/home/hmo/MoFin/gateway/logs/rec_digest_queue.jsonl' if not _os.path.exists(qf): return 0 @@ -1176,14 +1193,47 @@ def flush_rec_digest(max_items=5): _os.remove(qf) items.sort(key=lambda x: x.get('rr') or 0, reverse=True) top = items[:max_items] - lines = [f"📈 新增推荐 {len(items)} 只(按RR排序,精选前{len(top)}):"] - for it in top: + + # ── 现金预算(决定操盘建议)── + cash_note = "" + try: + conn = _sq.connect("/home/hmo/MoFin/data/mofin.db") + r = conn.execute("SELECT cash, total_assets FROM portfolio_summary WHERE id=1").fetchone() + conn.close() + if r and r[1]: + cash, total = r[0] or 0, r[1] + budget_pct = cash / total * 100 + cum = 0.0 + buys = [] + for it in top: + import re as _re + m = _re.search(r'(\d+(?:\.\d+)?)\s*%', it.get('position') or '') + pct = float(m.group(1)) if m else 8.0 + if cum + pct <= budget_pct + 1e-9: + buys.append(f"{it.get('name') or it['code']}≈{pct:.0f}%") + cum += pct + cash_note = (f"现金{cash/10000:.1f}万({budget_pct:.1f}%)|按预算本次可执行: " + + ("、".join(buys) if buys else "无(预算不足,先排队观察)") + + (f"(合计≈{cum:.0f}%)" if buys else "")) + except Exception: + pass + + lines = [f"📈 新增推荐 {len(items)} 只(按RR排序):"] + for i, it in enumerate(top): lines.append(f"• {it.get('name') or it['code']}({it['code']}) {it['signal']}" f" 区{it.get('entry_low') or '—'}~{it.get('entry_high') or '—'}" f" 损{it.get('stop_loss') or '—'} 盈{it.get('take_profit') or '—'}" - f" RR={it.get('rr') or 0}") + f" RR={it.get('rr') or 0} 仓位{it.get('position') or '—'}") + # 头部 2 只附策略依据(12维全文节选) + if i < 2: + if it.get('strategy_excerpt'): + lines.append(f" 依据: {it['strategy_excerpt'][:300]}") + elif it.get('full_analysis'): + lines.append(f" 依据: {it['full_analysis'][:400]}") if len(items) > max_items: lines.append(f"…另有 {len(items) - max_items} 只详见盯盘推荐操作区") + if cash_note: + lines.append("💰 " + cash_note) try: import sys as _s, os as _o2 _s.path.insert(0, '/home/hmo/MoFin/deploy/profile-scripts')