From a40b97f5ca01dc2e0a4a7c1b69279249d9edf288 Mon Sep 17 00:00:00 2001 From: hmo Date: Mon, 20 Jul 2026 08:48:08 +0800 Subject: [PATCH] feat(analysis): systematic daily 12-dim LLM analysis for holdings+watchlist MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Gap (reported by user via zhiwei): premarket full review updated technical params but full_analysis (12-dim LLM matrix) was empty for new holdings and stale for old ones — batch_reassess existed but was never wired into the daily pipeline and only covered watchlist. System fix: - premarket_full_review.py: new Step 1.5 runs batch_reassess --type holding --today every trading day 08:10 (force-refresh today's analysis, timeout 3600s, result in summary.json) - batch_reassess.py: - coverage: --type holding|watchlist|all (was watchlist-only) - staleness: analysis >20h stale gets refreshed (was: skip if any analysis exists = forever stale) - --today flag: force re-analyze if not reassessed since 04:00 today - cash/total read live from portfolio_summary (was hardcoded 321271/ 952879 from weeks ago) - HK stock prefix fix (5-digit codes -> hk, was sending sz00700) - watchlist_12d_backfill.py: wrapper for hermes cron (no args support) - cron job '批量补全九维分析-一次性' -> '自选12维分析补全-每日午间' (daily 12:30 weekdays, covers 109 watchlist stocks missing analysis) Verified: 300308 got 1920-char 12-dim analysis written to DB at 08:41, signal=观望, stop/take-profit updated. --- deploy/profile-scripts/batch_reassess.py | 91 +++++++++++++++---- .../profile-scripts/premarket_full_review.py | 32 ++++++- .../profile-scripts/watchlist_12d_backfill.py | 13 +++ scripts/check_backfill_job.py | 7 ++ scripts/check_db_state.py | 25 +++++ scripts/sc3.py | 9 ++ scripts/update_backfill_cron.py | 27 ++++++ scripts/verify_300308.py | 7 ++ 8 files changed, 189 insertions(+), 22 deletions(-) create mode 100644 deploy/profile-scripts/watchlist_12d_backfill.py create mode 100644 scripts/check_backfill_job.py create mode 100644 scripts/check_db_state.py create mode 100644 scripts/sc3.py create mode 100644 scripts/update_backfill_cron.py create mode 100644 scripts/verify_300308.py diff --git a/deploy/profile-scripts/batch_reassess.py b/deploy/profile-scripts/batch_reassess.py index 196dab1b..b201c6e8 100644 --- a/deploy/profile-scripts/batch_reassess.py +++ b/deploy/profile-scripts/batch_reassess.py @@ -1,9 +1,14 @@ #!/usr/bin/env python3 -"""batch_reassess.py — 批量补全九维分析（逐只处理，间隔防限流） +"""batch_reassess.py — 批量补全12维(九维矩阵)LLM分析（逐只处理，间隔防限流） -用法: python3 batch_reassess.py [--all] [--code XXXXXX] +用法: + python3 batch_reassess.py # 所有缺分析/过期的 active 策略 + python3 batch_reassess.py --type holding # 只处理持仓策略 + python3 batch_reassess.py --type watchlist # 只处理自选策略 + python3 batch_reassess.py --type holding --today # 持仓每日刷新（今早未评过的强制重评） + python3 batch_reassess.py --code XXXXXX # 单只 -流程：收集最新数据 → 调LLM(gateway)写九维分析+策略 → 保存到DB +流程：收集最新数据 → 调LLM(gateway)写12维分析+策略 → 保存到DB """ import sys, json, subprocess, sqlite3, re, time from datetime import datetime @@ -11,9 +16,10 @@ from datetime import datetime DB = "/home/hmo/MoFin/data/mofin.db" GATEWAY = "http://127.0.0.1:8643/v1/chat/completions" COOLDOWN_HOURS = 1 +STALE_HOURS = 20 # 分析超过20小时视为过期，需要重评 def has_llm_analysis(code): - """检查是否为LLM生成的九维分析（>500字）""" + """检查是否为LLM生成的12维分析（>500字）""" conn = sqlite3.connect(DB) r = conn.execute("SELECT LENGTH(full_analysis) FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone() conn.close() @@ -33,6 +39,34 @@ def in_cooldown(code): except: return False +def analysis_stale(code, force_today=False): + """分析是否过期（>STALE_HOURS 或 force_today 时今早4点前未重评）""" + conn = sqlite3.connect(DB) + r = conn.execute("SELECT reassessed_at FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone() + conn.close() + if not r or not r[0]: + return True + try: + last = datetime.fromisoformat(r[0]) + if force_today: + today4am = datetime.now().replace(hour=4, minute=0, second=0, microsecond=0) + return last < today4am + return (datetime.now() - last).total_seconds() / 3600 > STALE_HOURS + except: + return True + +def get_portfolio(): + """从 portfolio_summary 读实时现金/总资产（不再硬编码）""" + try: + conn = sqlite3.connect(DB) + r = conn.execute("SELECT cash, total_assets FROM portfolio_summary WHERE id=1").fetchone() + conn.close() + if r and r[1]: + return int(r[0] or 0), int(r[1]) + except Exception: + pass + return 0, 0 + def collect_data(code): """收集最新数据""" data = {"code": code} @@ -55,7 +89,14 @@ def collect_data(code): conn.close() # 从腾讯API拉最新价和基本面 - prefix = "sh" if str(code).startswith(("6","9")) else "sz" + # 代码前缀：5位=港股(hk)，6/9开头=沪(sh)，其他=深(sz) + _c = str(code) + if len(_c) == 5: + prefix = "hk" + elif _c.startswith(("6", "9")): + prefix = "sh" + else: + prefix = "sz" try: r = subprocess.run(["curl", "-s", f"http://qt.gtimg.cn/q={prefix}{code}"], capture_output=True, timeout=10) parts = r.stdout.decode("gbk", errors="ignore").split("~") @@ -81,8 +122,9 @@ def collect_data(code): def build_prompt(data): """构建LLM prompt，要求输出完整策略""" - cash = 321271 # 可用现金（从DB读取） - total = 952879 # 总资产 + cash, total = get_portfolio() # 实时从 portfolio_summary 读 + if not total: + cash, total = 241330, 929727 # 兜底（DB读不到时） # 拉取资金流数据 _flow_note = "暂无资金流数据" @@ -271,20 +313,21 @@ def save_result(code, full_text, parsed): conn.close() -def process_stock(code): +def process_stock(code, force_today=False): """处理单只股票""" print(f"\n{'='*50}") print(f"处理: {code}") print(f"{'='*50}") - if has_llm_analysis(code): - print(f" ⏭ 已有LLM九维分析，跳过") - return False - if in_cooldown(code): print(f" ⏭ 冷却期内，跳过") return False + # 有分析且未过期 → 跳过（除非 force_today 且今早未评） + if has_llm_analysis(code) and not analysis_stale(code, force_today): + print(f" ⏭ 已有12维分析且未过期，跳过") + return False + print(f" 收集数据...", flush=True) data = collect_data(code) if not data.get("price"): @@ -329,29 +372,41 @@ def process_stock(code): def main(): codes = [] + force_today = "--today" in sys.argv + dtype = None + if "--type" in sys.argv: + idx = sys.argv.index("--type") + dtype = sys.argv[idx + 1] # holding | watchlist | all if "--code" in sys.argv: idx = sys.argv.index("--code") codes = [sys.argv[idx+1]] else: - # 所有自选策略 + # 按类型筛选 active 策略 + type_map = {"holding": "持仓策略", "watchlist": "自选策略"} conn = sqlite3.connect(DB) - rows = conn.execute("SELECT code FROM holding_strategies WHERE status='active' AND decision_type='自选策略' ORDER BY code").fetchall() + if dtype in type_map: + rows = conn.execute( + "SELECT code FROM holding_strategies WHERE status='active' AND decision_type=? ORDER BY code", + (type_map[dtype],)).fetchall() + else: + rows = conn.execute( + "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)}只") + print(f"待处理: {len(codes)}只 (type={dtype or 'all'}, force_today={force_today})") ok = 0 fail = 0 skip = 0 for i, code in enumerate(codes): - if has_llm_analysis(code): - print(f" [{i+1}/{len(codes)}] ⏭ {code} 已有LLM分析") + if has_llm_analysis(code) and not analysis_stale(code, force_today): + print(f" [{i+1}/{len(codes)}] ⏭ {code} 已有12维分析且未过期") skip += 1 continue print(f" [{i+1}/{len(codes)}] ", end="", flush=True) - if process_stock(code): + if process_stock(code, force_today): ok += 1 else: fail += 1 diff --git a/deploy/profile-scripts/premarket_full_review.py b/deploy/profile-scripts/premarket_full_review.py index 1cebcd1c..2a8a9319 100644 --- a/deploy/profile-scripts/premarket_full_review.py +++ b/deploy/profile-scripts/premarket_full_review.py @@ -2,16 +2,17 @@ """premarket_full_review.py — 盘前全量重评 执行顺序： -1. regenerate_all() 全量技术分析重评（持仓+自选） -2. watchlist_auto_exit() 自选退出检查 -3. 输出摘要 +1. regenerate_all() 全量技术参数重评（持仓+自选） +2. batch_reassess.py --type holding --today 持仓12维LLM分析（每日强制刷新） +3. watchlist_auto_exit() 自选退出检查 +4. 输出摘要 调度：交易日 08:10（A股09:30开盘） """ import sys, os, json sys.path.insert(0, '/home/hmo/MoFin') -# Step 1: 全量重评 +# Step 1: 全量技术参数重评 print("=" * 50) print("📊 盘前全量重评开始") print("=" * 50) @@ -19,6 +20,28 @@ from strategy_lifecycle import regenerate_all result = regenerate_all(stdout=True) print(f"\n重评完成: {result.get('ok',0)}/{result.get('total',0)}成功") +# Step 1.5: 持仓 12 维 LLM 深度分析（每日强制，14只约8-10分钟） +print("\n" + "=" * 50) +print("🧠 持仓12维LLM分析（每日强制刷新）") +print("=" * 50) +import subprocess as _sp +analysis_result = {"ok": 0, "fail": 0, "skip": 0} +try: + r = _sp.run( + ["python3", "/home/hmo/.hermes/profiles/position-analyst/scripts/batch_reassess.py", + "--type", "holding", "--today"], + capture_output=True, text=True, timeout=3600) + print(r.stdout[-2000:] if len(r.stdout) > 2000 else r.stdout) + if r.returncode != 0 and r.stderr: + print(f"⚠️ stderr: {r.stderr[:300]}") + # 从输出尾部解析统计 + import re as _re + m = _re.search(r"完成: (\d+)成功, (\d+)失败, (\d+)跳过", r.stdout) + if m: + analysis_result = {"ok": int(m.group(1)), "fail": int(m.group(2)), "skip": int(m.group(3))} +except Exception as e: + print(f"⚠️ 12维分析步骤异常: {e}") + # Step 2: 自选退出 print("\n" + "=" * 50) print("🔍 自选退出检查") @@ -30,6 +53,7 @@ exited = auto_exit(dry_run=False) summary = { "premarket_at": __import__('datetime').datetime.now().isoformat(), "reassess": result, + "llm_analysis_12d": analysis_result, "auto_exit": [{"code": c, "name": n, "reason": r} for c, n, s, r in exited], "total_kept": result.get('total', 0) - len(exited), } diff --git a/deploy/profile-scripts/watchlist_12d_backfill.py b/deploy/profile-scripts/watchlist_12d_backfill.py new file mode 100644 index 00000000..b95a9ad9 --- /dev/null +++ b/deploy/profile-scripts/watchlist_12d_backfill.py @@ -0,0 +1,13 @@ +#!/usr/bin/env python3 +"""watchlist_12d_backfill.py — 自选12维LLM分析补全（每日午间） + +包装 batch_reassess.py --type watchlist，供 hermes cron 调度（不支持参数）。 +调度：交易日 12:30（午间低峰，自选补全约80分钟） +""" +import subprocess, sys + +r = subprocess.run( + [sys.executable, "/home/hmo/MoFin/deploy/profile-scripts/batch_reassess.py", + "--type", "watchlist"], + timeout=7200) +sys.exit(r.returncode) \ No newline at end of file diff --git a/scripts/check_backfill_job.py b/scripts/check_backfill_job.py new file mode 100644 index 00000000..b217a471 --- /dev/null +++ b/scripts/check_backfill_job.py @@ -0,0 +1,7 @@ +import json +d = json.load(open('/home/hmo/.hermes/profiles/position-analyst/cron/jobs.json')) +jobs = d if isinstance(d, list) else d.get('jobs', []) +for j in jobs: + if '批量补全' in j.get('name', '') or '九维' in j.get('name', ''): + print(json.dumps({k: j.get(k) for k in ('name', 'script', 'no_agent', 'prompt', 'schedule', 'enabled')}, + ensure_ascii=False, indent=2)) \ No newline at end of file diff --git a/scripts/check_db_state.py b/scripts/check_db_state.py new file mode 100644 index 00000000..7e1695af --- /dev/null +++ b/scripts/check_db_state.py @@ -0,0 +1,25 @@ +import sqlite3 +conn = sqlite3.connect('/home/hmo/MoFin/data/mofin.db') +print('=== portfolio_summary schema ===') +for r in conn.execute("SELECT sql FROM sqlite_master WHERE name='portfolio_summary'"): + print(r[0]) +print() +print('=== latest row ===') +cur = conn.execute('SELECT * FROM portfolio_summary ORDER BY id DESC LIMIT 1') +cols = [d[0] for d in cur.description] +row = cur.fetchone() +for c, v in zip(cols, row): + print(f' {c} = {v}') +print() +print('=== decision_type distribution ===') +for r in conn.execute("SELECT decision_type, COUNT(*) FROM holding_strategies WHERE status='active' GROUP BY decision_type"): + print(f' {r[0]}: {r[1]}') +print() +print('=== full_analysis staleness ===') +for r in conn.execute(""" + SELECT decision_type, + SUM(CASE WHEN full_analysis IS NULL OR LENGTH(full_analysis)<500 THEN 1 ELSE 0 END) as missing, + SUM(CASE WHEN LENGTH(full_analysis)>=500 THEN 1 ELSE 0 END) as has_fa, + COUNT(*) as total + FROM holding_strategies WHERE status='active' GROUP BY decision_type"""): + print(f' {r[0]}: missing={r[1]} has={r[2]} total={r[3]}') \ No newline at end of file diff --git a/scripts/sc3.py b/scripts/sc3.py new file mode 100644 index 00000000..bc386a1d --- /dev/null +++ b/scripts/sc3.py @@ -0,0 +1,9 @@ +import ast, sys +for p in ['/home/hmo/MoFin/deploy/profile-scripts/batch_reassess.py', + '/home/hmo/MoFin/deploy/profile-scripts/premarket_full_review.py']: + try: + ast.parse(open(p).read()) + print('OK:', p) + except SyntaxError as e: + print('SYNTAX ERROR:', p, e) + sys.exit(1) \ No newline at end of file diff --git a/scripts/update_backfill_cron.py b/scripts/update_backfill_cron.py new file mode 100644 index 00000000..bfef2ff4 --- /dev/null +++ b/scripts/update_backfill_cron.py @@ -0,0 +1,27 @@ +import json, shutil +from datetime import datetime + +jf = '/home/hmo/.hermes/profiles/position-analyst/cron/jobs.json' +shutil.copy(jf, jf + '.bak-20260720') + +d = json.load(open(jf)) +jobs = d if isinstance(d, list) else d.get('jobs', []) +for j in jobs: + if j.get('name') == '批量补全九维分析-一次性': + j['name'] = '自选12维分析补全-每日午间' + j['script'] = 'watchlist_12d_backfill.py' + j['schedule'] = {"kind": "cron", "expr": "30 12 * * 1-5", "display": "30 12 * * 1-5"} + j['schedule_display'] = "30 12 * * 1-5" + j['next_run_at'] = "2026-07-20T12:30:00+08:00" + j['state'] = 'scheduled' + print('updated job:', j['name'], '| script:', j['script'], '| schedule:', j['schedule_display']) + break +else: + print('job not found!') + +if isinstance(d, list): + json.dump(jobs, open(jf, 'w'), ensure_ascii=False, indent=2) +else: + d['jobs'] = jobs + json.dump(d, open(jf, 'w'), ensure_ascii=False, indent=2) +print('saved') \ No newline at end of file diff --git a/scripts/verify_300308.py b/scripts/verify_300308.py new file mode 100644 index 00000000..b85a8c43 --- /dev/null +++ b/scripts/verify_300308.py @@ -0,0 +1,7 @@ +import sqlite3 +conn = sqlite3.connect('/home/hmo/MoFin/data/mofin.db') +r = conn.execute("SELECT code, LENGTH(full_analysis), reassessed_at, timing_signal, stop_loss, take_profit FROM holding_strategies WHERE code='300308'").fetchone() +print(r) +# also show first 300 chars of the analysis +r2 = conn.execute("SELECT substr(full_analysis, 1, 400) FROM holding_strategies WHERE code='300308'").fetchone() +print(r2[0] if r2 else 'none') \ No newline at end of file