From a559fc84dd05a119c0fd47f8ccecadd41b49fb8e Mon Sep 17 00:00:00 2001
From: hmo <staymo7777@gmail.com>
Date: Fri, 24 Jul 2026 09:02:34 +0800
Subject: [PATCH] =?UTF-8?q?feat:=20=E6=89=B9=E9=87=8F=E9=87=8D=E8=AF=84?=
 =?UTF-8?q?=E5=88=86=E7=89=87=E5=B9=B6=E5=8F=91=EF=BC=88=E8=80=81=E7=88=B8?=
 =?UTF-8?q?=EF=BC=9A=E8=BF=99=E4=B9=88=E5=A4=9Akey=E4=B8=8D=E8=83=BD?=
 =?UTF-8?q?=E5=B9=B6=E5=8F=91=EF=BC=9F=EF=BC=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- batch_reassess --shard K/N: index%N==K分片,N进程互不重叠
- llm_client LLM_KEY_OFFSET: key池起始位错开,避免N进程压同一首选key
- SKIP_FLUSH=1时worker不发摘要,parallel_batch.sh收尾统一flush
- parallel_batch.sh: N路并行+等待+统一digest
---
 deploy/profile-scripts/batch_reassess.py | 26 +++++++++++++++++-------
 deploy/profile-scripts/llm_client.py     | 11 +++++++---
 deploy/profile-scripts/parallel_batch.sh | 21 +++++++++++++++++++
 3 files changed, 48 insertions(+), 10 deletions(-)
 create mode 100644 deploy/profile-scripts/parallel_batch.sh

diff --git a/deploy/profile-scripts/batch_reassess.py b/deploy/profile-scripts/batch_reassess.py
index 0774b134..bff4a798 100644
--- a/deploy/profile-scripts/batch_reassess.py
+++ b/deploy/profile-scripts/batch_reassess.py
@@ -633,8 +633,18 @@ def main():
                 "SELECT code FROM holding_strategies WHERE status='active' ORDER BY decision_type, code").fetchall()
         conn.close()
         codes = [r[0] for r in rows]
-    
-    print(f"待处理: {len(codes)}只 (type={dtype or 'all'}, force_today={force_today})")
+
+    # ── 分片并发（2026-07-24 老爸：这么多key不能并发？）──
+    # --shard K/N：本 worker 只处理 index%N==K 的股票，N 个进程并发互不重叠。
+    _shard_k, _shard_n = 0, 1
+    if "--shard" in sys.argv:
+        _sk = sys.argv[sys.argv.index("--shard") + 1]  # 格式 K/N
+        _shard_k, _shard_n = int(_sk.split("/")[0]), int(_sk.split("/")[1])
+    if _shard_n > 1:
+        codes = [c for i, c in enumerate(codes) if i % _shard_n == _shard_k]
+
+    print(f"待处理: {len(codes)}只 (type={dtype or 'all'}, force_today={force_today}"
+          + (f", shard={_shard_k}/{_shard_n}" if _shard_n > 1 else "") + ")")
     
     ok = 0
     fail = 0
@@ -679,11 +689,13 @@ def main():
     print(f"完成: {ok}成功, {fail}失败, {skip}跳过")
     print(f"{'='*50}")
     # ── 推荐摘要：本轮新增推荐聚成一条推送（防逐只轰炸）──
-    try:
-        from mofin_db import flush_rec_digest
-        flush_rec_digest()
-    except Exception as _e:
-        print(f"  ⚠️ 推荐摘要发送失败: {_e}")
+    # 并发分片模式（SKIP_FLUSH=1）下由 launcher 统一 flush，避免先到者发半成品摘要
+    if not os.environ.get("SKIP_FLUSH"):
+        try:
+            from mofin_db import flush_rec_digest
+            flush_rec_digest()
+        except Exception as _e:
+            print(f"  ⚠️ 推荐摘要发送失败: {_e}")
 
 if __name__ == "__main__":
     main()
diff --git a/deploy/profile-scripts/llm_client.py b/deploy/profile-scripts/llm_client.py
index d006c1f9..1f3dface 100644
--- a/deploy/profile-scripts/llm_client.py
+++ b/deploy/profile-scripts/llm_client.py
@@ -107,14 +107,19 @@ _KEY_POOL_TS = 0
 
 
 def _get_key_pool():
-    """key 池缓存 5 分钟。"""
+    """key 池缓存 5 分钟。LLM_KEY_OFFSET 环境变量：并发 worker 起始 key 错开，
+    避免 N 个进程同时压同一个首选 key（2026-07-24 并发分片配套）。"""
     global _KEY_POOL, _KEY_POOL_TS
-    import time as _t
+    import time as _t, os as _os
     if not _KEY_POOL or (_t.time() - _KEY_POOL_TS) > 300:
         _KEY_POOL = _load_key_pool()
         _KEY_POOL_TS = _t.time()
         if _KEY_POOL:
-            print(f"  [LLM] key池: {[k for k, _ in _KEY_POOL]}", flush=True)
+            _off = int(_os.environ.get("LLM_KEY_OFFSET", "0") or 0)
+            if _off and len(_KEY_POOL) > 1:
+                _off = _off % len(_KEY_POOL)
+                _KEY_POOL = _KEY_POOL[_off:] + _KEY_POOL[:_off]
+            print(f"  [LLM] key池: {[k for k, _ in _KEY_POOL]}" + (f" (offset={_off})" if _off else ""), flush=True)
     return _KEY_POOL
 
 
diff --git a/deploy/profile-scripts/parallel_batch.sh b/deploy/profile-scripts/parallel_batch.sh
new file mode 100644
index 00000000..4a44bcd0
--- /dev/null
+++ b/deploy/profile-scripts/parallel_batch.sh
@@ -0,0 +1,21 @@
+#!/bin/bash
+# parallel_batch.sh — 并发批量12维重评（分片+key池偏移+统一flush）
+# 用法: bash parallel_batch.sh [N] [type]
+#   N=worker数(默认4)  type=holding|watchlist|all(默认all)
+N=${1:-4}
+TYPE=${2:-all}
+cd /home/hmo/MoFin/deploy/profile-scripts
+echo "[parallel] launching $N workers (type=$TYPE) at $(date '+%F %T')"
+pids=()
+for ((k=0; k<N; k++)); do
+  SKIP_FLUSH=1 LLM_KEY_OFFSET=$k python3 batch_reassess.py --type "$TYPE" --shard "$k/$N" \
+    > "/tmp/reassess_parallel_w$k.log" 2>&1 &
+  pids+=($!)
+  echo "[parallel] worker $k pid=${pids[$k]}"
+done
+for p in "${pids[@]}"; do wait "$p"; done
+echo "[parallel] all workers done at $(date '+%F %T')"
+# 统一推荐摘要（防各worker半成品）
+python3 -c "from mofin_db import flush_rec_digest; flush_rec_digest()"
+echo "[parallel] digest flushed"
+grep -h "^完成:" /tmp/reassess_parallel_w*.log