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 "/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