feat: 批量重评分片并发（老爸：这么多key不能并发？）

- 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

@@ -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()

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
 
 

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